A crowd in one chat

The chat's own model answers, then each other member in order, then the order runs
backwards asking each whether it disagrees, ending at the main model, which either
closes or sends them round again. Design and reasoning: LLeMbas.wiki/Crowd-chats.

THE SPEAKER SEAM, WHICH IS ALSO A BUG FIX

`chat_service.speaker_for` makes the *message* name the answering model and the
chat only the default. That closes a live half-wired feature -- `wake_chat` takes a
model override and `schedule/runner` passes one, and it reached the row and never
the request, so a schedule naming another model got the chat's model wearing the
other one's name.

The seam is wider than `build_request`: `{{model_name}}`, the authored prompt's
model layer, `vision` (where a wrong answer makes the endpoint reject the whole
request), the effort vocabulary (which raises inside the model's own chat template,
and whose refusal narrows every Model row sharing the id), `resolve_tools`,
`context_length` -> `_too_big`, and `ToolContext.model_id`. `resolve_endpoint` may
now only write back `chat.connection_id` when the speaker *is* the chat's model.

WHY N CHAINED REPLIES

`Generation` is one reply's state and `_follow` streams per message, so one
generation cannot stream into nine bubbles and `ensure` would not know which of the
nine it was after a restart. A subagent per speaker cannot work either: its answer
comes back as a tool result and tool results are never replayed, so speaker 3 could
not see speaker 2 -- which is the whole point. Chained, exactly one incomplete row
exists at a time, and `tests/test_crowd_chain.py` asserts that at every
observation.

The round lives on `Message.crowd_json`, not on the chat: the row is the authority,
and chat-level state would describe turns a rewind or a restart had removed.
`crowd.next_turn` is pure, so all eight refusals are tested with no endpoint.

THREE RULES, EACH A BUG WRITTEN THE OTHER WAY ROUND

- `if not _advance_crowd(g): _drain(g)` -- advancing must *suppress* draining, or a
  queued human turn puts a second incomplete row beside the next speaker's.
- `_advance_crowd` refuses unless the finishing row is the newest, or regenerating
  member 2 creates a second member 3 and two chains race down one turn.
- an error skips one speaker and two in a row end the round: the usual failure is a
  small member's window overflowing, and `_drain`'s stop-on-error would kill every
  crowd at whichever member is smallest.

Each other speaker's turn is relabelled as attributed user content, which is both
how a model can disagree with words it did not write and how the history keeps
alternating. The per-speaker instruction is payload-only -- as a row it could be
dropped from the request by a `created_at` tie, and every later speaker would answer
it. Compaction, titling and the notification are gated to once per turn; `_inject`
is off during a round; the way back gets no tools and a member is treated as
unattended.

Membership stores the model as text with no foreign key: "Test & refresh" deletes
and recreates Model rows, and a cascade would empty the crowd out of every chat.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

CHANGELOG.md

@@ -16,6 +16,39 @@ for 1.0.0 have something to be assembled from.
 
 ## Unreleased
 
+## 1.6.0
+
+- **A chat can have a crowd.** Switch it on under Admin → Agents, and each chat's
+  settings panel offers the other models. The chat's own model answers first, then
+  each of the others in turn; then the order runs **backwards**, each one asked
+  whether it disagrees with anything said; and it ends back at the first model,
+  which either writes the final answer or sends them round again. Every
+  contribution is its own bubble with its own avatar, its own metrics and a chip
+  saying which speaker it is and which pass it belongs to.
+
+  What it costs is stated where you turn it on and again where you pick the
+  models, because it is easy to underestimate: one turn is **models × rounds ×
+  2 − 1** replies, so four models over two rounds is fifteen. On a single local
+  endpoint every change of speaker also loads a different model. Your own warning
+  is built into the defaults — larger crowds start going round in circles — so the
+  round limit is two, and it is a limit ordinary work will reach rather than a
+  runaway backstop.
+
+  Details worth knowing: each model sees the others' answers **quoted and
+  attributed**, never as its own words, so it can actually disagree with them; a
+  member you can no longer reach is skipped and said so rather than silently
+  dropped; a member whose endpoint fails is skipped, and two failures in a row end
+  the round; **Stop ends the round**, not just the model writing at the time; and
+  a message typed during a round waits for the round rather than interleaving with
+  it. Every sentence a crowd sends is editable under Admin → Prompts.
+
+- Fixed: **a schedule that named its own model was ignored.** It was written on
+  the reply and never sent, so the bubble showed the model you chose while the
+  answer came from the chat's model. The same fix makes the crowd possible: the
+  reply itself now says which model is answering, rather than the conversation
+  deciding for all of them. Regenerating somebody's turn in a crowd keeps that
+  model rather than silently switching to the chat's.
+
 ## 1.5.0
 
 - **A model's personality is now yours, not the instance's.** Each account gets

src/lembas/__init__.py

@@ -1,3 +1,3 @@
 """LLeMbas - a Middle-earth themed web UI for OpenAI-compatible LLM endpoints."""
 
-__version__ = "1.5.0"
+__version__ = "1.6.0"

src/lembas/api/admin_agents.py

@@ -66,6 +66,10 @@ async def agents_page(request: Request, db: Db, user: AdminUser, saved: bool = F
             # reply is allowed to set going on its own, and a nav entry for one
             # card would be worse than the near-miss.
             "subagents": settings_store.subagents(db),
+            # And a third group on the same page, for the same reason: a crowd is
+            # not an agent-chat feature either, but this is where somebody comes to
+            # find out what one turn is allowed to set going.
+            "crowd": settings_store.crowd(db),
             "saved": saved,
         },
     )
@@ -111,6 +115,35 @@ async def save_subagents(
     return RedirectResponse("/admin/agents?saved=1", status_code=status.HTTP_303_SEE_OTHER)
 
 
+@router.post("/crowd")
+async def save_crowd(
+    db: Db,
+    user: AdminUser,
+    enabled: bool = Form(False),
+    max_models: int = Form(4),
+    max_rounds: int = Form(2),
+    wall_seconds: int = Form(900),
+    collapse_agreement: bool = Form(False),
+) -> Response:
+    """Its own route, for the reason `save_subagents` gives above."""
+    settings_store.update(
+        db,
+        {
+            "enabled": enabled,
+            # Clamped here as well as on read. Every floor is one: a zero would be
+            # the feature switched off wearing the switch's clothes, and that is a
+            # thing to answer in one place.
+            "max_models": min(max(max_models, 1), 8),
+            "max_rounds": min(max(max_rounds, 1), 5),
+            "wall_seconds": min(max(wall_seconds, 60), 7200),
+            "collapse_agreement": collapse_agreement,
+        },
+        key=settings_store.CROWD,
+    )
+    log.info("crowd %s by %s", "enabled" if enabled else "disabled", user.email)
+    return RedirectResponse("/admin/agents?saved=1", status_code=status.HTTP_303_SEE_OTHER)
+
+
 @router.post("")
 async def save_agents(
     db: Db,

src/lembas/api/chats.py

@@ -1453,6 +1453,32 @@ def _queue_frames(
                 + "</div>"
             )
 
+    # The next speaker of a crowd round, on the same frame and by the same
+    # mechanism -- an incomplete assistant bubble carries `sse-connect`, so htmx
+    # opens the next stream itself and there is no new streaming machinery here at
+    # all.
+    #
+    # Its own branch and not the one above, deliberately. That one also re-renders
+    # "the last user turn at or before this bubble" to take Send now and Discard
+    # off it, and a crowd has no queued user turn: the swap would either re-render
+    # a node that was already correct or target one that is not in the document,
+    # where htmx silently does nothing. A branch that sometimes does nothing is a
+    # branch nobody can reason about.
+    if getattr(generation, "crowded", False):
+        following = list(
+            db.scalars(
+                select(Message)
+                .where(Message.chat_id == chat.id, Message.complete.is_(False))
+                .order_by(Message.created_at, Message.id)
+            )
+        )
+        for speaker_row in following:
+            out_of_band.append(
+                '<div hx-swap-oob="beforeend:#thread">'
+                + _render_bubble(db, chat, owner, speaker_row)
+                + "</div>"
+            )
+
     return "".join(moved), "".join(out_of_band)
 
 
@@ -2089,6 +2115,42 @@ async def update_chat(request: Request, db: Db, user: RequiredUser, chat_id: str
             else []
         )
 
+    if "crowd_model_ids" in form:
+        # The same shape as the bases above: one field always sent, so clearing
+        # every box clears the crowd. Checked against what this person can reach
+        # rather than against what exists, or the picker is advisory and a crafted
+        # request walks past it -- the reasoning the model branch carries.
+        from lembas.db.models import CrowdMember
+
+        settings = settings_store.crowd(db)
+        reachable = {
+            model.model_id for model in chat_service.available_models(db, user)
+        }
+        wanted: list[str] = []
+        for value in form.getlist("crowd_model_ids"):
+            value = str(value).strip()
+            # Never the chat's own model: it would answer twice in a row, which is
+            # nobody's idea of a second opinion.
+            if value and value in reachable and value != chat.model_id and value not in wanted:
+                wanted.append(value)
+        wanted = wanted[: int(settings["max_models"])]
+
+        chat.crowd = [
+            CrowdMember(
+                model_id=model_id,
+                connection_id=next(
+                    (
+                        model.connection_id
+                        for model in chat_service.available_models(db, user)
+                        if model.model_id == model_id
+                    ),
+                    None,
+                ),
+                position=index,
+            )
+            for index, model_id in enumerate(wanted)
+        ]
+
     submitted_params = {name: form[name] for name in _PARAM_RANGES if name in form}
     if submitted_params:
         if not allowed.get("chat.params"):
@@ -2228,7 +2290,14 @@ async def regenerate(
     message.content = ""
     message.error = ""
     message.complete = False
-    message.model_id = chat.model_id
+    # Whose reply this was stays whose reply it is, unless the chat's model has
+    # been changed since -- in which case regenerating is how somebody asks for
+    # the new one. Before 1.6.0 this always reset to the chat's model, which was
+    # merely a wrong label; now that the row *is* the model that answers, it would
+    # silently regenerate somebody else's turn as the chat's model.
+    if not (message.model_id or "").strip():
+        message.model_id = chat.model_id
+        message.connection_id = chat.connection_id
     _note_rewind(chat)
     db.commit()
     # restart, not ensure: this is the one caller that reuses a Message row, and

src/lembas/api/pages.py

@@ -82,6 +82,7 @@ def _chat_context(db: DBSession, user: User, chat: Chat | None) -> dict:
             else []
         ),
         "attached_base_ids": [base.id for base in chat.knowledge_bases] if chat else [],
+        **_crowd_context(db, user, chat, models),
         # What *this* model takes, not the three every model used to be assumed
         # to take. The vocabulary is per model -- gpt-oss has no `xhigh` and
         # Bonsai has no `high`, and sending the wrong one does not degrade, it
@@ -195,6 +196,42 @@ def _scope_context(db: DBSession, user: User, chat: Chat | None) -> dict:
     }
 
 
+def _crowd_context(db: DBSession, user: User, chat: Chat | None, models: list) -> dict:
+    """Who else could answer in this chat, and what that would cost.
+
+    Empty — and the panel then shows nothing rather than an empty control — when
+    the feature is off, when there is nobody else to add, or on the new-chat
+    screen, where there is no chat to attach anybody to yet.
+
+    The cost is spelled out because it is the thing somebody will not have thought
+    about: a turn is `speakers x rounds x 2 - 1` replies, and on one local endpoint
+    each change of speaker is also a model load.
+    """
+    from lembas.services import crowd as crowd_service
+
+    settings = settings_store.crowd(db)
+    if chat is None or not settings["enabled"]:
+        return {"crowd_available": [], "crowd_member_ids": [], "crowd_skipped": []}
+
+    others = [model for model in models if model.model_id != chat.model_id]
+    members = [
+        row.model_id
+        for row in sorted(chat.crowd, key=lambda row: (row.position, row.model_id))
+    ]
+    reachable = {model.model_id for model in others}
+    speakers = 1 + len([model_id for model_id in members if model_id in reachable])
+    rounds = int(settings["max_rounds"])
+    return {
+        "crowd_available": others,
+        "crowd_member_ids": [model_id for model_id in members if model_id in reachable],
+        "crowd_skipped": crowd_service.unreachable_members(db, chat, user),
+        # One round is out and back: everybody answers, everybody but the last is
+        # asked whether they disagree, and the main model closes.
+        "crowd_replies": max(1, speakers * 2 - 1),
+        "crowd_rounds": rounds,
+    }
+
+
 # What a gate is called in the menu. A gate covers several tools, so no single
 # tool's label is the right name for it.
 _GATE_LABELS = {

src/lembas/db/models/__init__.py

@@ -31,6 +31,7 @@ from lembas.db.models.chat import (
     ROLE_TOOL,
     ROLE_USER,
     Chat,
+    CrowdMember,
     Folder,
     Message,
 )
@@ -163,6 +164,7 @@ __all__ = [
     "Report",
     "Schedule",
     "Chat",
+    "CrowdMember",
     "Job",
     "Connection",
     "CustomTool",

src/lembas/db/models/chat.py

@@ -5,7 +5,15 @@ from __future__ import annotations
 from datetime import datetime
 from typing import TYPE_CHECKING, Any
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+)
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 from lembas.db.base import Base, Timestamps, UUIDPrimaryKey
@@ -309,10 +317,60 @@ class Chat(UUIDPrimaryKey, Timestamps, Base):
         "KnowledgeBase", secondary="chat_knowledge_bases"
     )
 
+    # The other models answering in this chat, in the order they speak. Empty is
+    # every chat that has ever existed: one model, answering on its own.
+    crowd: Mapped[list[CrowdMember]] = relationship(
+        back_populates="chat",
+        cascade="all, delete-orphan",
+        order_by="CrowdMember.position",
+    )
+
     def __repr__(self) -> str:
         return f"<Chat {self.title!r}>"
 
 
+class CrowdMember(UUIDPrimaryKey, Timestamps, Base):
+    """One extra model answering in a chat, and where it sits in the order.
+
+    A row rather than an association table because it carries an order and has
+    nothing to associate *to*:
+
+    🚨 **the model is stored as text, with no foreign key to `models`.** "Test &
+    refresh" on the connection screen deletes every model the endpoint has
+    stopped listing and creates it again when it comes back, so a foreign key
+    with `ON DELETE CASCADE` -- which is what copying `chat_knowledge_bases`
+    would have given -- means one refresh taken while an endpoint happened to be
+    loading something else silently empties the crowd out of every chat, with no
+    row left to explain it. This is the reasoning `Chat.model_id`,
+    `ssh_profile_id` and `compacted_through_id` all carry, and the same trap that
+    lost the image reviewer its model in 1.4.x.
+
+    A member that no longer resolves is therefore skipped at send time and shown
+    struck through, rather than being deleted by something nobody asked.
+
+    `connection_id` is nullable and usually empty, meaning "resolve it from the
+    id"; it matters only where two connections offer the same model, since their
+    capabilities and effort lists are separate rows.
+    """
+
+    __tablename__ = "chat_crowd"
+    __table_args__ = (UniqueConstraint("chat_id", "model_id"),)
+
+    chat_id: Mapped[str] = mapped_column(
+        String(32), ForeignKey("chats.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    model_id: Mapped[str] = mapped_column(String(300), nullable=False)
+    connection_id: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    # Where this member speaks. The chat's own model is always first and is not a
+    # row here, so these start at 1 in spirit and are only ever compared.
+    position: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
+
+    chat: Mapped[Chat] = relationship(back_populates="crowd")
+
+    def __repr__(self) -> str:
+        return f"<CrowdMember {self.model_id} at {self.position}>"
+
+
 class Message(UUIDPrimaryKey, Timestamps, Base):
     __tablename__ = "messages"
 
@@ -340,14 +398,46 @@ class Message(UUIDPrimaryKey, Timestamps, Base):
     # Milliseconds spent producing the reasoning, for the "Thought for Xs" label.
     reasoning_ms: Mapped[int] = mapped_column(Integer, default=0, nullable=False)
 
+    # Which model wrote this, or is about to. Written on every assistant
+    # placeholder at creation and, from 1.6.0, **read back as the model that
+    # answers** -- `chat_service.speaker_for`. Before that it was a display
+    # snapshot only, and the two could disagree: `wake_chat` accepts a model
+    # override that reached this column and never reached the request, so a
+    # schedule naming another model got the chat's model wearing this label.
     model_id: Mapped[str] = mapped_column(String(300), default="")
 
+    # Which connection that model was reached through. Nullable and usually
+    # empty, meaning "resolve it from the model id as this application always
+    # has"; it matters only where the same id is offered by two connections,
+    # since `Model` is unique on the pair and their capabilities, context lengths
+    # and effort lists are separate rows.
+    #
+    # No foreign key, deliberately, and the same reasoning `Chat.model_id`
+    # carries: a transcript has to survive an administrator deleting a
+    # connection, and `migrations.py` compiles only the column type -- so a
+    # REFERENCES clause would exist on a fresh database and not on an upgraded
+    # one. Validated on read instead.
+    connection_id: Mapped[str | None] = mapped_column(String(32), nullable=True)
+
     # What the model did before answering: one entry per tool call, with its
     # arguments and results. Shown in the transcript so the sources behind an
     # answer stay visible, and deliberately NOT replayed as context on the next
     # turn -- see services/generation.py for why.
     tool_calls_json: Mapped[list[Any]] = mapped_column(JSONList, default=list)
 
+    # Where this message sits in a crowd round: the turn it belongs to, the
+    # round, the phase, and which speaker it is. NULL on every message that is
+    # not part of one, which is every message this application has ever written
+    # before 1.6.0.
+    #
+    # On the row and not on the chat, deliberately. "The row is the authority,
+    # not the registry" is the rule the reload story was won with, and round
+    # state on the chat reintroduces the split it was won against: a restart
+    # between speakers, or a rewind that deletes these rows, would leave
+    # chat-level state describing turns that no longer exist -- which is the
+    # problem `compacted_through_id` already documents.
+    crowd_json: Mapped[dict[str, Any] | None] = mapped_column(JSONDict, nullable=True)
+
     # Where each round's contribution ended, so `content`, `reasoning` and
     # `tool_calls_json` can be shown as the one sequence they actually were
     # rather than as three stacked zones. One entry per closed step, holding the

src/lembas/services/chat.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import logging
 import re
+from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
@@ -46,19 +47,71 @@ TITLE_MAX_TOKENS = 512
 TEMPORARY_LIFETIME = timedelta(hours=24)
 
 
-def resolve_endpoint(db: DBSession, chat: Chat) -> tuple[Endpoint, str]:
+@dataclass(frozen=True)
-    """Find the connection and model a chat should use.
+class Speaker:
+    """Which model is answering one reply, and through which connection.
+
+    The pair and not the id, because `Model` is unique on
+    `(connection_id, model_id)`: the same name can live behind two endpoints and
+    an id alone does not say which. `images/tool.py:_reviewer` already resolves a
+    model this way.
+
+    Frozen, and passed rather than re-derived, for the reason `Endpoint` is a
+    snapshot: a generation outlives the request that started it, and "who is
+    answering" must not be able to change underneath a reply that is already
+    streaming.
+    """
+
+    model_id: str
+    connection_id: str | None = None
+
+
+def speaker_for(db: DBSession, chat: Chat, message: Message | None = None) -> Speaker:
+    """Who is answering: the message being written into, or else the chat.
+
+    **The row names the model and the chat is only the default.** Until 1.6.0 the
+    answering model was `chat.model_id` and nothing else, while `Message.model_id`
+    was written on every placeholder and read only for display -- so the bubble's
+    avatar and the request could disagree, and did: `wake_chat` accepts a
+    `model_id` override and `schedule/runner` passes `schedule.model_id or
+    chat.model_id`, which reached the row and never reached the request. A
+    schedule naming another model got the chat's model wearing the other one's
+    name.
+
+    Reading it off the row is also what makes a reply survive a restart, because
+    `_follow` calls `ensure`, which starts a *new* generation against the same
+    row -- so anything the request depends on has to be durable, and the registry
+    is not. This is the rule the reload story was won with: the row is the
+    authority.
+    """
+    if message is not None and (message.model_id or "").strip():
+        return Speaker(message.model_id, getattr(message, "connection_id", None) or None)
+    return Speaker(chat.model_id, chat.connection_id)
+
+
+def resolve_endpoint(
+    db: DBSession, chat: Chat, speaker: Speaker | None = None
+) -> tuple[Endpoint, str]:
+    """Find the connection and model a reply should use.
 
     Chats store the model id as text rather than a foreign key so history
     survives an admin deleting a connection, which means the mapping back to a
     live connection has to be resolved at send time and can legitimately fail.
+
+    `speaker` defaults to the chat's own model, so every existing caller behaves
+    exactly as it did.
     """
-    if not chat.model_id:
+    speaker = speaker or speaker_for(db, chat)
+    if not speaker.model_id:
         raise LLMError("This chat has no model selected.")
+    # Whether resolving a fallback may be *written back* to the chat. It may only
+    # when the speaker is the chat's own model: a crowd member or a schedule's
+    # model finding its way to another connection must not repoint the chat.
+    speaks_for_chat = speaker.model_id == chat.model_id
 
     connection: Connection | None = None
-    if chat.connection_id:
+    if speaker.connection_id:
-        connection = db.get(Connection, chat.connection_id)
+        connection = db.get(Connection, speaker.connection_id)
 
     if connection is None or not connection.enabled:
         # The original connection is gone or disabled. Any enabled connection
@@ -67,7 +120,7 @@ def resolve_endpoint(db: DBSession, chat: Chat) -> tuple[Endpoint, str]:
             select(Model)
             .join(Connection)
             .where(
-                Model.model_id == chat.model_id,
+                Model.model_id == speaker.model_id,
                 Model.enabled.is_(True),
                 Connection.enabled.is_(True),
             )
@@ -76,13 +129,14 @@ def resolve_endpoint(db: DBSession, chat: Chat) -> tuple[Endpoint, str]:
         if model is None:
             raise LLMError(
                 f"No enabled connection currently offers the model "
-                f"'{chat.model_id}'. Pick another model for this chat."
+                f"'{speaker.model_id}'. Pick another model for this chat."
             )
         connection = model.connection
-        chat.connection_id = connection.id
+        if speaks_for_chat:
-        db.commit()
+            chat.connection_id = connection.id
+            db.commit()
 
-    return Endpoint.from_connection(connection), chat.model_id
+    return Endpoint.from_connection(connection), speaker.model_id
 
 
 def document_context(message: Message) -> str:
@@ -190,7 +244,9 @@ def folder_system_prompt(db: DBSession, chat: Chat) -> str:
     return ""
 
 
-def effective_system_prompt(db: DBSession, chat: Chat) -> str:
+def effective_system_prompt(
+    db: DBSession, chat: Chat, speaker: Speaker | None = None
+) -> str:
     """The system prompt a chat actually runs with.
 
     Four layers, most specific wins outright:
@@ -214,9 +270,9 @@ def effective_system_prompt(db: DBSession, chat: Chat) -> str:
     if inherited := folder_system_prompt(db, chat):
         return inherited
 
-    model = db.scalar(
+    # The *answering* model's layer, which is not always the chat's: a crowd
-        select(Model).where(Model.model_id == chat.model_id).order_by(Model.position)
+    # member speaking in somebody else's chat brings its own prompt with it.
-    )
+    model = model_row(db, speaker or Speaker(chat.model_id, chat.connection_id))
     if model is not None and (model.system_prompt or "").strip():
         return model.system_prompt.strip()
 
@@ -230,6 +286,7 @@ def build_messages(
     upto: Message | None = None,
     vision: bool = False,
     system_prompt: str | None = None,
+    speaker: Speaker | None = None,
 ) -> list[dict]:
     """Assemble the message list to send upstream.
 
@@ -302,24 +359,196 @@ def build_messages(
             continue
         payload.append(message_payload(message, vision=vision))
 
+    if speaker is not None:
+        payload = _as_one_speaker_sees_it(db, payload, history, speaker, upto=upto)
+
     return payload
 
 
-def model_for(db: DBSession, chat: Chat) -> Model | None:
+def _as_one_speaker_sees_it(
-    """The Model row a chat is using, or None if it has gone.
+    db: DBSession,
+    payload: list[dict[str, Any]],
+    history: list[Message],
+    speaker: Speaker,
+    *,
+    upto: Message | None = None,
+) -> list[dict[str, Any]]:
+    """Rewrite a crowd transcript from one speaker's point of view.
+
+    Two problems, one pass.
+
+    **Another speaker's reply must not arrive as this one's own prior turn.** Sent
+    verbatim, every assistant message in the payload reads as something *this*
+    model said -- so it defends sentences it never wrote, and cannot disagree with
+    them, which is the whole point of the backward pass. Each other speaker's turn
+    is therefore relabelled as user content behind a fragment-driven "«Label»
+    said:".
+
+    **Consecutive assistant turns break strict-alternation chat templates**, which
+    this project already knows: `task.compact_ack` exists so a compacted history
+    still alternates, and several templates reject one that does not. Relabelling
+    fixes that by construction, and the adjacent user turns it creates are merged.
+
+    ⚠ The relabelled entry is built here rather than by calling `message_payload`
+    with a swapped role. That function attaches image parts when the role is
+    `user` and the model has vision, so a swapped assistant turn carrying a
+    generated image would silently become a multimodal list -- and an endpoint
+    that rejects one rejects every later turn with it.
+    """
+    from lembas.services import prompts as prompts_service
+
+    # Nothing to do for the ordinary case: one model, and every assistant turn in
+    # the payload is its own.
+    others = {
+        message.model_id
+        for message in history
+        if message.role == ROLE_ASSISTANT
+        and (message.model_id or "")
+        and message.model_id != speaker.model_id
+    }
+    if not others:
+        return payload
+
+    labels = {
+        model_id: (row.label if (row := model_row(db, Speaker(model_id))) else model_id)
+        for model_id in others
+    }
+    template = prompts_service.resolve(db, "crowd.said") or "{{crowd_speaker}} answered:"
+
+    # The payload and the history line up only over the message rows: the system
+    # turn and a compaction pair come first and belong to nobody. Walking from the
+    # end is what pairs them without counting.
+    rows = [
+        message
+        for message in history
+        if not (upto is not None and message.id == upto.id)
+    ]
+    rewritten: list[dict[str, Any]] = []
+    for index, entry in enumerate(payload):
+        row = None
+        offset = index - (len(payload) - len(rows))
+        if 0 <= offset < len(rows):
+            row = rows[offset]
+        if (
+            row is not None
+            and entry.get("role") == ROLE_ASSISTANT
+            and (row.model_id or "") in others
+        ):
+            lead = template.replace("{{crowd_speaker}}", labels[row.model_id])
+            body = entry.get("content")
+            rewritten.append(
+                {"role": ROLE_USER, "content": f"{lead}\n\n{body if isinstance(body, str) else ''}"}
+            )
+            continue
+        rewritten.append(entry)
+
+    return _merge_user_turns(rewritten)
+
+
+def _with_crowd_instruction(
+    db: DBSession, payload: list[dict[str, Any]], turn, *, again: bool
+) -> list[dict[str, Any]]:
+    """Append what this speaker has been asked to do, as the closing user turn.
+
+    🚨 **Payload only. No row is written for it.** Writing the instruction into the
+    transcript the way `wake_chat` writes a background job's turn was the first
+    design and is wrong three times over. `build_messages` orders history by
+    `created_at` alone and `break`s at the placeholder, so on a shared microsecond
+    the placeholder sorts first and the instruction is dropped from the request
+    entirely -- the hazard `thread_tail` already carries an explicit tiebreak for.
+    It would double the rows in a turn, all of them bubbles somebody has to scroll
+    past. And every later speaker would read the previous speaker's instruction as
+    an ordinary user turn and answer that too.
+
+    The compaction summary is inserted the same way and for the same reason: a
+    turn in the payload with nothing behind it (`build_messages`).
+    """
+    from lembas.services import crowd as crowd_service
+    from lembas.services import prompts as prompts_service
+
+    if turn.phase == crowd_service.PHASE_OUT:
+        key = "crowd.turn"
+    elif turn.phase == crowd_service.PHASE_BACK:
+        key = "crowd.disagree"
+    else:
+        # Two fragments, not one with a clause in it: inviting a choice the model
+        # cannot express is worse than not offering it, and a model without the
+        # tools capability has no `crowd_again` to call.
+        key = "crowd.close" if again else "crowd.close_final"
+
+    text = (prompts_service.resolve(db, key) or "").strip()
+    if not text:
+        # Cleared on purpose is the administrator switching this wording off, and
+        # an empty user turn is not a thing to send.
+        return payload
+    return _merge_user_turns([*payload, {"role": ROLE_USER, "content": text}])
+
+
+def _merge_user_turns(payload: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Fold adjacent user turns into one, so the history still alternates.
+
+    Only where both are plain strings: a turn carrying content parts is a
+    multimodal message and joining one to a string would destroy it.
+    """
+    merged: list[dict[str, Any]] = []
+    for entry in payload:
+        last = merged[-1] if merged else None
+        if (
+            last is not None
+            and last.get("role") == ROLE_USER
+            and entry.get("role") == ROLE_USER
+            and isinstance(last.get("content"), str)
+            and isinstance(entry.get("content"), str)
+        ):
+            merged[-1] = {
+                **last,
+                "content": f"{last['content']}\n\n{entry['content']}",
+            }
+            continue
+        merged.append(entry)
+    return merged
+
+
+def model_row(db: DBSession, speaker: Speaker) -> Model | None:
+    """The Model row a speaker names, or None if it has gone.
 
     Looked up by id rather than held as a foreign key, for the same reason
     resolve_endpoint does: chats store the model as text so history survives an
-    administrator deleting a connection.
+    administrator deleting a connection. The connection narrows it when one is
+    named, because two connections may offer the same id and their capabilities,
+    context length and effort lists are separate rows.
     """
+    if not speaker.model_id:
+        return None
+    if speaker.connection_id:
+        exact = db.scalar(
+            select(Model).where(
+                Model.model_id == speaker.model_id,
+                Model.connection_id == speaker.connection_id,
+            )
+        )
+        if exact is not None:
+            return exact
     return db.scalar(
-        select(Model).where(Model.model_id == chat.model_id).order_by(Model.position)
+        select(Model).where(Model.model_id == speaker.model_id).order_by(Model.position)
     )
 
 
-def model_supports(db: DBSession, chat: Chat, capability: str) -> bool:
+def model_for(db: DBSession, chat: Chat) -> Model | None:
-    """Whether the chat's current model is marked as having a capability."""
+    """The Model row a chat is using. The display answer; see `model_row`."""
-    model = model_for(db, chat)
+    return model_row(db, Speaker(chat.model_id, chat.connection_id))
+
+
+def model_supports(
+    db: DBSession, chat: Chat, capability: str, speaker: Speaker | None = None
+) -> bool:
+    """Whether the answering model is marked as having a capability.
+
+    ⚠ Worth getting right per speaker rather than per chat: `vision` decides
+    whether image parts go into the body, and an endpoint sent an image by a
+    model that cannot take one rejects **the whole request**, not the image.
+    """
+    model = model_row(db, speaker) if speaker is not None else model_for(db, chat)
     return bool(model and (model.capabilities_json or {}).get(capability))
 
 
@@ -331,12 +560,21 @@ def build_request(
     tools: list[dict[str, Any]] | None = None,
     user=None,
     force_tool: str = "",
+    speaker: Speaker | None = None,
+    crowd_turn=None,
+    crowd_again: bool = False,
 ) -> dict[str, Any]:
     """The whole request body, tools and harness included.
 
     Composed here rather than in the generation loop so that "what gets sent"
     has one answer, and so the harness cannot be forgotten by a future caller
     that offers tools.
+
+    `speaker` is who is answering; it defaults to the chat's own model, so a
+    caller that does not care behaves exactly as it did. Everything that differs
+    per model is resolved from it and not from the chat: the model name sent, the
+    vision decision, the authored prompt's model layer, `{{model_name}}`, the
+    personality, and the reasoning-effort vocabulary.
     """
     from lembas.services import harness as harness_service
     from lembas.services import prompts as prompts_service
@@ -346,10 +584,15 @@ def build_request(
         for key, value in (chat.params_json or {}).items()
         if key in FORWARDED_PARAMS and value not in (None, "")
     }
+    speaker = speaker or speaker_for(db, chat, upto)
+    if crowd_turn is None and upto is not None:
+        from lembas.services import crowd as crowd_service
+
+        crowd_turn = crowd_service.state_of(upto)
     # Images are only sent to a model an administrator has marked as having
     # vision. Sending them to one that has not is not a graceful degradation:
     # most endpoints reject the whole request.
-    vision = model_supports(db, chat, "vision")
+    vision = model_supports(db, chat, "vision", speaker=speaker)
 
     if user is None:
         from lembas.db.models import User
@@ -360,18 +603,23 @@ def build_request(
     # behaviour. See services/harness.py for why these are joined rather than
     # being two competing layers.
     system = harness_service.join(
-        harness_service.compose(db, user, tools, chat),
+        harness_service.compose(db, user, tools, chat, speaker=speaker),
-        effective_system_prompt(db, chat),
+        effective_system_prompt(db, chat, speaker),
         lead=prompts_service.render(db, "seam.authored_lead", {}),
     )
 
     body: dict[str, Any] = {
-        "model": chat.model_id,
+        "model": speaker.model_id,
         "messages": build_messages(
-            db, chat, upto=upto, vision=vision, system_prompt=system
+            db, chat, upto=upto, vision=vision, system_prompt=system, speaker=speaker
         ),
         **params,
     }
+    if crowd_turn is not None:
+        body["messages"] = _with_crowd_instruction(
+            db, body["messages"], crowd_turn, again=crowd_again
+        )
+
     if tools:
         body["tools"] = tools
         # Making the model call one particular tool, for `/image` -- the whole
@@ -388,14 +636,22 @@ def build_request(
         ):
             body["tool_choice"] = {"type": "function", "function": {"name": force_tool}}
 
-    # The model's own vocabulary, looked up here rather than passed in: every
+    # The *answering* model's own vocabulary, looked up here rather than passed
-    # caller of `build_request` would otherwise have to remember, which is the
+    # in: every caller of `build_request` would otherwise have to remember, which
-    # trap `audio_service.template_flags` fell into.
+    # is the trap `audio_service.template_flags` fell into.
-    chat_model = model_for(db, chat)
+    #
+    # ⚠ Per speaker and not per chat, and this one is not cosmetic: the
+    # vocabularies genuinely differ -- gpt-oss takes low/medium/high, a Bonsai
+    # takes low/medium/xhigh and *raises inside its chat template* on high -- so
+    # a chat's effort handed to another model fails the whole reply rather than
+    # being ignored. `_learn_refused_effort` then narrows every Model row sharing
+    # that id, so getting this wrong would also corrupt other models' lists as a
+    # side effect.
+    speaking_model = model_row(db, speaker)
     apply_effort(
         body,
         (chat.params_json or {}).get("reasoning_effort"),
-        efforts_for(chat_model) if chat_model is not None else None,
+        efforts_for(speaking_model) if speaking_model is not None else None,
     )
     return body
 

src/lembas/services/crowd.py

@@ -0,0 +1,383 @@
+"""Several models answering one turn, in order, then again in reverse.
+
+The shape the owner asked for: the chat's own model answers, then each other
+member in order; then the order runs **backwards**, each member asked whether it
+disagrees with anything; and it ends at the main model, which decides whether to
+go round again or stop.
+
+## Why N chained replies and not one clever one
+
+One `Generation` per speaker, one `Message` per speaker, chained where `_drain`
+already chains a queued turn. That is not the cheapest shape, it is the only one
+in which every existing invariant keeps holding for the reason it already holds:
+
+* `Generation` is **one reply's** state and `_follow` streams **per message**,
+  keyed on `generation.message_id`. One generation cannot stream into nine
+  bubbles without a second streaming protocol, and `ensure(chat_id, message_id)`
+  would have no answer to "which of the nine am I" after a restart.
+* Exactly one incomplete assistant row exists at any moment, so
+  `_reply_in_flight` needs no teaching and the composer queues for the whole
+  round.
+* Each speaker gets its own `steps_json`, `usage_json` and `model_id`, so the
+  avatar, the metrics chip and the regenerate button are per speaker with no new
+  rendering.
+
+A subagent per speaker was rejected outright: a helper is handed a *serialisation*
+of the conversation, its answer comes back as a tool result, and tool results are
+never replayed -- so speaker 3 could not see speaker 2, which is the entire point
+of a crowd. That feature already exists and is called `ask_friend`.
+
+## Where the round lives
+
+On the **message row**, in `Message.crowd_json`, and not on the chat. "The row is
+the authority, not the registry" is the rule the reload story was won with, and
+round state on the chat reintroduces exactly the split it was won against: a
+restart between speakers, or a rewind that deletes the rows, would leave
+chat-level state describing turns that no longer exist -- which is the problem
+`Chat.compacted_through_id` already documents.
+
+`Message.parent_id` is **not** used for grouping. It is reserved for conversation
+branching and says so in its own comment.
+
+## The scheduler is a pure function
+
+`next_turn` takes numbers and returns numbers. Every refusal -- out of rounds, out
+of time, nobody to ask, not the newest message -- is therefore testable without an
+endpoint, which matters because the refusals are the interesting half.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, replace
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session as DBSession
+
+from lembas.db.models import Chat, Message
+
+log = logging.getLogger(__name__)
+
+# The forward pass: everybody answers in order.
+PHASE_OUT = "out"
+# The way back: each member is asked whether it disagrees, in reverse order,
+# stopping one short of the main model.
+PHASE_BACK = "back"
+# The main model's last word, where it decides whether to go round again.
+PHASE_CLOSE = "close"
+
+PHASES = (PHASE_OUT, PHASE_BACK, PHASE_CLOSE)
+
+# Why a round ended, when it ended for a reason rather than by finishing.
+STOPPED_ROUNDS = "rounds"
+STOPPED_TIME = "time"
+STOPPED_ERRORS = "errors"
+
+# How many speaker errors in a row end the round. One is skipped: the commonest
+# failure in a crowd is not a dead endpoint but a small member's context window
+# overflowing on a transcript several models have been writing into, and killing
+# the round at whichever member is smallest is the wrong answer. Two in a row is
+# an endpoint that has actually gone, which is what `_drain`'s refusal protects
+# against and is worth keeping.
+MAX_CONSECUTIVE_ERRORS = 2
+
+
+@dataclass(frozen=True)
+class Turn:
+    """Where one crowd round has got to, as it is stored on a message."""
+
+    turn: str
+    round: int
+    phase: str
+    index: int
+    of: int
+    started_at: str
+    errors: int = 0
+    stopped: str = ""
+
+    def as_json(self) -> dict[str, Any]:
+        return {
+            "turn": self.turn,
+            "round": self.round,
+            "phase": self.phase,
+            "index": self.index,
+            "of": self.of,
+            "started_at": self.started_at,
+            "errors": self.errors,
+            "stopped": self.stopped,
+        }
+
+    @property
+    def is_main(self) -> bool:
+        return self.index == 0
+
+
+def state_of(message: Message | None) -> Turn | None:
+    """The round state on a message, or None if it is not part of one."""
+    raw = getattr(message, "crowd_json", None) or None
+    if not raw or not isinstance(raw, dict):
+        return None
+    try:
+        return Turn(
+            turn=str(raw.get("turn") or ""),
+            round=int(raw.get("round") or 1),
+            phase=str(raw.get("phase") or PHASE_OUT),
+            index=int(raw.get("index") or 0),
+            of=int(raw.get("of") or 1),
+            started_at=str(raw.get("started_at") or ""),
+            errors=int(raw.get("errors") or 0),
+            stopped=str(raw.get("stopped") or ""),
+        )
+    except (TypeError, ValueError):  # pragma: no cover - a hand-edited row
+        return None
+
+
+def now_stamp() -> str:
+    return datetime.now(UTC).isoformat()
+
+
+def elapsed(started_at: str) -> float:
+    """Seconds since a round began, or 0.0 if the stamp is unreadable.
+
+    Unreadable reads as "no time has passed" rather than as "out of time": a
+    round abandoned because of a bad timestamp would be a feature failing for a
+    reason nobody could see.
+    """
+    try:
+        began = datetime.fromisoformat(started_at)
+    except (TypeError, ValueError):
+        return 0.0
+    if began.tzinfo is None:
+        began = began.replace(tzinfo=UTC)
+    return max(0.0, (datetime.now(UTC) - began).total_seconds())
+
+
+def next_turn(
+    *,
+    speakers: int,
+    state: Turn | None,
+    turn_id: str,
+    again: bool = False,
+    errored: bool = False,
+    max_rounds: int = 2,
+    wall_seconds: int = 900,
+) -> Turn | None:
+    """Who speaks next, or None when the round is over.
+
+    Pure: numbers in, numbers out, no session and no clock beyond the stamp it is
+    handed. `speakers` counts the main model as one of them.
+
+    `state=None` means the reply that has just finished was the ordinary first
+    one, started by the composer as it always is -- so this is where a round
+    begins rather than continues.
+    """
+    if speakers < 2:
+        return None
+
+    if state is None:
+        return Turn(
+            turn=turn_id,
+            round=1,
+            phase=PHASE_OUT,
+            index=1,
+            of=speakers,
+            started_at=now_stamp(),
+        )
+
+    # Errors are counted consecutively, so one member timing out is skipped and
+    # an endpoint that has gone ends the round.
+    errors = state.errors + 1 if errored else 0
+    if errors >= MAX_CONSECUTIVE_ERRORS:
+        return replace(state, stopped=STOPPED_ERRORS)
+
+    if wall_seconds and elapsed(state.started_at) >= wall_seconds:
+        return replace(state, errors=errors, stopped=STOPPED_TIME)
+
+    carry = {
+        "turn": state.turn,
+        "of": speakers,
+        "started_at": state.started_at,
+        "errors": errors,
+    }
+
+    if state.phase == PHASE_OUT:
+        if state.index + 1 <= speakers - 1:
+            return Turn(round=state.round, phase=PHASE_OUT, index=state.index + 1, **carry)
+        # The forward pass is done. The way back starts one short of the speaker
+        # that has just finished -- asking it whether it disagrees with itself is
+        # a round spent on nothing.
+        if speakers - 2 >= 1:
+            return Turn(round=state.round, phase=PHASE_BACK, index=speakers - 2, **carry)
+        return Turn(round=state.round, phase=PHASE_CLOSE, index=0, **carry)
+
+    if state.phase == PHASE_BACK:
+        if state.index - 1 >= 1:
+            return Turn(round=state.round, phase=PHASE_BACK, index=state.index - 1, **carry)
+        return Turn(round=state.round, phase=PHASE_CLOSE, index=0, **carry)
+
+    # The main model has had its last word. Another round only if it asked for
+    # one *and* there is one left.
+    if not again:
+        return None
+    if state.round + 1 > max_rounds:
+        return replace(state, errors=errors, stopped=STOPPED_ROUNDS)
+    return Turn(round=state.round + 1, phase=PHASE_OUT, index=1, **carry)
+
+
+# --- Resolving the membership --------------------------------------------------
+def member_speakers(db: DBSession, chat: Chat, user=None) -> list:
+    """Every member that can actually be reached, in order, main model first.
+
+    Filtered through `permissions.models_visible_to` by way of
+    `chat_service.roster_models`, so a member whose access has been revoked, whose
+    model has been disabled, or whose row has gone is skipped rather than
+    attempted -- and the skip is visible in the transcript rather than silent.
+
+    Deduplicated against the main model: adding the chat's own model to the crowd
+    would have it answer twice in a row, which is not what anybody meant by it.
+    """
+    from lembas.services import chat as chat_service
+
+    reachable = {
+        model.model_id: model for model in chat_service.roster_models(db, user, exclude="")
+    }
+    speakers = [chat_service.Speaker(chat.model_id, chat.connection_id)]
+    seen = {chat.model_id}
+    for member in sorted(chat.crowd, key=lambda row: (row.position, row.model_id)):
+        if member.model_id in seen or member.model_id not in reachable:
+            continue
+        seen.add(member.model_id)
+        speakers.append(chat_service.Speaker(member.model_id, member.connection_id))
+    return speakers
+
+
+def unreachable_members(db: DBSession, chat: Chat, user=None) -> list[str]:
+    """Members that will be skipped, so a screen can say so rather than lie."""
+    from lembas.services import chat as chat_service
+
+    reachable = {
+        model.model_id for model in chat_service.roster_models(db, user, exclude="")
+    }
+    return [
+        member.model_id
+        for member in chat.crowd
+        if member.model_id not in reachable or member.model_id == chat.model_id
+    ]
+
+
+def is_newest(db: DBSession, message: Message) -> bool:
+    """Whether this is the last message in its chat.
+
+    The guard that stops a regenerate from forking the round. `restart` re-runs
+    `_run`, whose `finally` advances the crowd again -- and speakers further down
+    already exist, so without this, regenerating member 2 creates a second member
+    3 and two chains race down one turn. `_drain` never needed it, because a
+    queued row only ever exists *forward* of the reply.
+    """
+    latest = db.scalars(
+        select(Message)
+        .where(Message.chat_id == message.chat_id)
+        .order_by(Message.created_at.desc(), Message.id.desc())
+        .limit(1)
+    ).first()
+    return latest is not None and latest.id == message.id
+
+
+# --- Asking for another round ---------------------------------------------------
+async def _run_crowd_again(context, args: dict[str, Any]):
+    """Record that the main model wants the crowd to go round again.
+
+    Written onto the running `Generation` rather than onto the row, because it is
+    a fact about *this* reply and dies with it -- and onto a field rather than
+    parsed back out of the prose, for the reason `plan_json` exists: a sentinel
+    phrase in an answer is a decision nobody can see and a wording nobody can
+    change.
+
+    Offered only on the main model's closing turn and only while a round is left,
+    so a call arriving anywhere else is a call that was never on the table.
+    """
+    from lembas.services import generation as generation_service
+    from lembas.services.tools import ToolOutcome
+
+    reason = str(args.get("focus") or "").strip()
+    running = generation_service.running_for(context.chat_id) if context.chat_id else None
+    if running is None:
+        return ToolOutcome(
+            "There is no round to continue.",
+            {"name": "crowd_again", "status": "error", "error": "no round"},
+        )
+
+    running.crowd_again = True
+    return ToolOutcome(
+        "The others will answer again."
+        + (f" You have asked them to focus on: {reason}" if reason else "")
+        + " Finish your answer now: what you write is what the person reads for "
+        "this round.",
+        {
+            "name": "crowd_again",
+            "status": "ok",
+            "query": reason[:160],
+            "detail": "another round",
+        },
+    )
+
+
+def tool_defs() -> list:
+    """The one tool, offered only to the closing speaker of a crowd round."""
+    from lembas.services.tools import FAMILY_CROWD, RISK_READ, ToolDef
+
+    return [
+        ToolDef(
+            name="crowd_again",
+            family=FAMILY_CROWD,
+            description=(
+                "Send the other models round again, because the disagreement is "
+                "real and another pass would settle it. Say what they should focus "
+                "on. Use it sparingly: every round costs the person another wait, "
+                "and a crowd asked to go round because the discussion was "
+                "interesting will keep finding things to discuss. If the answers "
+                "have converged, or the disagreement is a matter of taste, or "
+                "nobody has said anything new on the way back, do not call this -- "
+                "write the answer instead."
+            ),
+            parameters={
+                "type": "object",
+                "properties": {
+                    "focus": {
+                        "type": "string",
+                        "description": (
+                            "What the next round should settle, in one sentence."
+                        ),
+                    }
+                },
+                "required": [],
+            },
+            run=_run_crowd_again,
+            # It changes nothing in the world; what it costs is more replies, and
+            # that is bounded by `crowd.max_rounds` rather than by an approval.
+            risk=RISK_READ,
+        ),
+    ]
+
+
+__all__ = [
+    "MAX_CONSECUTIVE_ERRORS",
+    "PHASES",
+    "PHASE_BACK",
+    "PHASE_CLOSE",
+    "PHASE_OUT",
+    "STOPPED_ERRORS",
+    "STOPPED_ROUNDS",
+    "STOPPED_TIME",
+    "Turn",
+    "elapsed",
+    "is_newest",
+    "member_speakers",
+    "next_turn",
+    "now_stamp",
+    "state_of",
+    "tool_defs",
+    "unreachable_members",
+]

src/lembas/services/generation.py

@@ -33,6 +33,7 @@ from lembas.security import permissions
 from lembas.services import canvas as canvas_service
 from lembas.services import chat as chat_service
 from lembas.services import compaction as compaction_service
+from lembas.services import crowd as crowd_service
 from lembas.services import interaction, settings_store, tokens, tool_labels
 from lembas.services import metrics as metrics_service
 from lembas.services import prompts as prompts_service
@@ -222,6 +223,15 @@ class Generation:
     # -- the one frame that reaches a browser after a reply is over.
     drained: bool = False
     injected_ids: list[str] = field(default_factory=list)
+    # A crowd round, seen from one speaker's side. `crowded` says this reply's
+    # ending handed the turn to the next speaker -- read by `_follow`, exactly as
+    # `drained` is, to put the next bubble on the `done` frame. `crowd_again` is
+    # the main model having called `crowd_again` on its closing turn: a field
+    # rather than a parse of the prose, for the reason `plan_json` exists, and
+    # on the generation rather than the row because it is a fact about this reply
+    # and dies with it.
+    crowded: bool = False
+    crowd_again: bool = False
     # Images this reply produced, waiting to be bound to its message row. The
     # runner writes the file and the `Attachment`; only `_persist` may say which
     # turn it belongs to, which is the same division of labour `canvas` above
@@ -599,7 +609,16 @@ async def _run(generation: Generation) -> None:
         # assembly path. Here rather than in post_message because that route's
         # whole contract is to return immediately, and a three-second
         # summarisation in front of it would break exactly that.
-        await _maybe_compact(generation)
+        # Once per turn, on the reply that opens it. Three reasons, and the
+        # first is the one that bites: `should_compact` reads `context_limit` off
+        # the *last complete* assistant turn's usage, which mid-crowd is the
+        # previous **speaker** -- so an 8k member at position three tells a 128k
+        # member at position four to compact. `last_complete`'s own promise that
+        # the cut lands on a reply and therefore leaves a history starting on a
+        # user turn is also false mid-round. And compacting during a round would
+        # ask the way back whether it disagrees with a summary of itself.
+        if _opens_the_turn_id(generation):
+            await _maybe_compact(generation)
 
         # Before the session opens, for the same reason compaction is: the
         # listing is an SSH round trip, and holding a database session across
@@ -614,7 +633,14 @@ async def _run(generation: Generation) -> None:
                 generation.error = "That chat no longer exists."
                 return
 
-            endpoint, model_id = chat_service.resolve_endpoint(db, chat)
+            # Who is answering, from the row being written into rather than
+            # from the chat. The row is durable and this generation is not: a
+            # restart turns `_follow` into `ensure`, which starts a brand new
+            # `_run` against the same message, and everything the request depends
+            # on has to survive that. It is also the only thing that can make the
+            # bubble's avatar and the model actually asked agree.
+            speaker = chat_service.speaker_for(db, chat, message)
+            endpoint, model_id = chat_service.resolve_endpoint(db, chat, speaker)
             owner = db.get(User, chat.user_id)
 
             # Before the request is built, not while it streams. Every other
@@ -631,13 +657,42 @@ async def _run(generation: Generation) -> None:
             # Resolved once, so that what the loop is allowed to *run* is the
             # same set the endpoint was *offered* -- not whatever happens to
             # exist by the time a call comes back.
-            toolset = tools_service.resolve_tools(db, chat, owner)
+            # Where this speaker sits in a crowd round, if it is in one. Read
+            # once, here, and used for three decisions: which tools it may have,
+            # which instruction closes its request, and whether it may ask for
+            # another round.
+            crowd_state = crowd_service.state_of(message)
+            crowd_settings = settings_store.crowd(db)
+            may_ask_again = bool(
+                crowd_state is not None
+                and crowd_state.phase == crowd_service.PHASE_CLOSE
+                and crowd_state.round < int(crowd_settings["max_rounds"])
+            )
+            toolset = tools_service.resolve_tools(
+                db, chat, owner, speaker, crowd_turn=crowd_state, crowd_again=may_ask_again
+            )
             offered = toolset.schemas
             payload = chat_service.build_request(
-                db, chat, upto=message, tools=offered, user=owner, force_tool=generation.force_tool
+                db,
+                chat,
+                upto=message,
+                tools=offered,
+                user=owner,
+                force_tool=generation.force_tool,
+                speaker=speaker,
+                crowd_turn=crowd_state,
+                # Asked of the resolved set rather than of the settings: a model
+                # without the tools capability gets no tools at all, so inviting it
+                # to call `crowd_again` would be offering a choice it cannot
+                # express -- and `crowd.close_final` is the wording for that.
+                crowd_again="crowd_again" in toolset.by_name,
             )
             question = _question_from(payload)
-            needs_title = not chat.title_generated
+            # Once per turn. A crowd member titling the chat would name it after
+            # `_question_from`'s last user turn, which under the crowd relabelling
+            # is another model's quoted answer -- so the chat gets called after a
+            # quotation. The main model's first reply is the one that titles.
+            needs_title = not chat.title_generated and _opens_the_turn(message)
             # An agent chat is titled from its opening words and never costs a
             # model call for it. That prompt is a good title already -- somebody
             # starting one states an objective, not a topic -- while an ordinary
@@ -654,16 +709,22 @@ async def _run(generation: Generation) -> None:
             # Read here, with the rest, because titling happens after this
             # session has closed and must not open another one.
             title_prompt = prompts_service.resolve(db, "task.title")
-            tool_context = tools_service.context_for(db, owner, chat, tools=toolset)
+            tool_context = tools_service.context_for(
+                db, owner, chat, tools=toolset, speaker=speaker
+            )
 
-            model = chat_service.model_for(db, chat)
+            # The answering model's window, not the chat's. `_too_big` is the one
+            # budget that stops a reply dead rather than asking it to wrap up, so
+            # judging a small model's request against a large model's ceiling is
+            # how a reply fails with no explanation in it.
+            model = chat_service.model_row(db, speaker)
             generation.context_limit = model.context_length if model is not None else 0
             # Kept for `_inject`, which builds a user turn after this session
             # has closed. A turn taken in mid-reply has to be shaped exactly as
             # the same words typed a moment later would have been -- images to a
             # vision model, a plain string to anything else, or the endpoint
             # rejects the whole request.
-            vision = chat_service.model_supports(db, chat, "vision")
+            vision = chat_service.model_supports(db, chat, "vision", speaker=speaker)
             # Resolved while the session is open, like everything else here.
             # Empty for an admin and for a user in no group, which is every
             # instance that has not set one -- see permissions.limits_for.
@@ -1082,7 +1143,18 @@ async def _run(generation: Generation) -> None:
         # `_persist` is: `_follow` breaks the instant it sees that flag, and the
         # frame it then sends is the one that has to carry the next turn's
         # bubbles. There is no push channel that outlives a single reply.
-        _drain(generation)
+        #
+        # 🚨 Advancing a crowd round *suppresses* the drain, and the order of this
+        # sentence is the whole of it. Written the other way round -- advance, then
+        # drain -- a queued human turn typed during a round would create a second
+        # incomplete assistant row beside the next speaker's, which is two
+        # generations in one chat: the state `_reply_in_flight`, `_too_many_replies`,
+        # `wake.lock_for` and the superseded guards in `_persist`/`_drain` all exist
+        # to make unreachable, and whose symptom is a Stop button pointing at
+        # whichever bubble comes first in the document. The queue waits for the
+        # round; that is what a queue is for.
+        if not _advance_crowd(generation):
+            _drain(generation)
         generation.done = True
         generation.finished_at = datetime.now(UTC)
         generation.touch()
@@ -2112,9 +2184,166 @@ def _drain(generation: Generation) -> None:
     generation.drained = True
 
 
+def _advance_crowd(generation: Generation) -> bool:
+    """Start the next speaker of a crowd round. True if one was started.
+
+    The imperative shell around `crowd.next_turn`, which is pure -- so everything
+    interesting about this (the eight ways a round declines to continue) is tested
+    without an endpoint, and what is left here is reading rows and writing one.
+
+    Three refusals of its own, and each is a bug if it is left out:
+
+    * **Superseded.** The same guard `_persist` and `_drain` carry: this reply is
+      no longer the one registered for its message.
+    * **Stopped.** A person pressing Stop ends the round, not just the speaker
+      writing at the time. `_drain` refuses after a stop for the same reason and
+      it is the same reason here -- somebody asked for it to end.
+    * **Not the newest message.** `regenerate` calls `restart`, whose `finally`
+      runs this again -- and the speakers after it already exist. Without this,
+      regenerating member 2 creates a second member 3 and two chains race down one
+      turn. `_drain` never needed the guard because a queued row only ever exists
+      *forward* of the reply.
+
+    An **error** does not end the round: `crowd.next_turn` counts consecutive
+    failures and abandons after two, because the commonest failure in a crowd is a
+    small member's context window overflowing rather than a dead endpoint, and
+    ending the round there would kill every crowd at whichever member is smallest.
+    """
+    owner = _RUNNING.get(generation.message_id)
+    if owner is not None and owner is not generation:
+        return False
+    if generation.stopped:
+        return False
+
+    try:
+        with session_scope() as db:
+            chat = db.get(Chat, generation.chat_id)
+            message = db.get(Message, generation.message_id)
+            if chat is None or message is None:
+                return False
+
+            settings = settings_store.crowd(db)
+            if not settings["enabled"] or not chat.crowd:
+                return False
+            if not crowd_service.is_newest(db, message):
+                return False
+
+            owner_user = db.get(User, chat.user_id)
+            speakers = crowd_service.member_speakers(db, chat, owner_user)
+            speakers = speakers[: int(settings["max_models"]) + 1]
+
+            state = crowd_service.state_of(message)
+            # The turn a round belongs to: the user message this all answers.
+            turn_id = state.turn if state is not None else _turn_anchor(db, message)
+            following = crowd_service.next_turn(
+                speakers=len(speakers),
+                state=state,
+                turn_id=turn_id,
+                again=generation.crowd_again,
+                errored=bool(generation.error),
+                max_rounds=int(settings["max_rounds"]),
+                wall_seconds=int(settings["wall_seconds"]),
+            )
+            if following is None:
+                return False
+            if following.stopped:
+                # Recorded on the row that ended it, so the transcript can say
+                # why a round stopped rather than simply stopping. Nothing else
+                # needs writing: there is no next speaker.
+                message.crowd_json = following.as_json()
+                db.commit()
+                return False
+
+            speaker = speakers[following.index]
+            placeholder = chat_service.create_message(
+                db,
+                chat,
+                ROLE_ASSISTANT,
+                "",
+                complete_=False,
+                model_id=speaker.model_id,
+            )
+            placeholder.connection_id = speaker.connection_id
+            placeholder.crowd_json = following.as_json()
+            db.commit()
+            chat_id, next_id = chat.id, placeholder.id
+    except Exception:  # noqa: BLE001 - the reply is over either way
+        log.exception("could not advance the crowd in chat %s", generation.chat_id)
+        return False
+
+    # Outside the session, like `_drain`: this starts a task.
+    ensure(chat_id, next_id)
+    generation.crowded = True
+    return True
+
+
+def _opens_the_turn(message: Message) -> bool:
+    """Whether this reply is the first one answering a question.
+
+    True for every ordinary reply, and for a crowd only for the main model's
+    opening turn -- which is the one with no crowd state on it at all, because a
+    round begins when that reply *finishes*.
+    """
+    return crowd_service.state_of(message) is None
+
+
+def _opens_the_turn_id(generation: Generation) -> bool:
+    """`_opens_the_turn` before the session is open, by message id.
+
+    `_maybe_compact` runs before `_run` reads anything, so this opens its own
+    session -- one primary-key lookup, and only on a chat that has a crowd.
+    """
+    try:
+        with session_scope() as db:
+            message = db.get(Message, generation.message_id)
+            return message is None or _opens_the_turn(message)
+    except Exception:  # noqa: BLE001 - compaction is best-effort anyway
+        return True
+
+
+def _ends_the_turn(message: Message) -> bool:
+    """Whether this reply is the last one the person is waiting for.
+
+    True for every ordinary reply, and for a crowd only on the main model's
+    closing turn. What is gated on it is everything that should happen once per
+    question rather than once per speaker: the unread dot, the web push, and the
+    chat's title.
+    """
+    state = crowd_service.state_of(message)
+    if state is None:
+        return True
+    return state.phase == crowd_service.PHASE_CLOSE
+
+
+def _turn_anchor(db, message: Message) -> str:
+    """The user turn a round answers, for a round that is only now beginning.
+
+    The last user message at or before this reply. Only read once per round -- it
+    is carried on every later turn's state -- and it exists so a rewind can tell
+    which rows belonged to which question.
+    """
+    row = db.scalars(
+        select(Message)
+        .where(
+            Message.chat_id == message.chat_id,
+            Message.role == ROLE_USER,
+            Message.created_at <= message.created_at,
+        )
+        .order_by(Message.created_at.desc(), Message.id.desc())
+        .limit(1)
+    ).first()
+    return row.id if row is not None else ""
+
+
 def _inject(generation: Generation, chat_id: str, vision: bool) -> dict | None:
     """Take the oldest waiting prompt into this reply, between two rounds.
 
+    ⚠ Never during a crowd round. This restamps the placeholder's `created_at` so
+    the reply sorts after the prompt it answers, which mid-round reorders the
+    speakers underneath themselves -- and the round's own bookkeeping counts an
+    anchor that has moved. The turn stays queued and arrives after the round as a
+    clean new question with a round of its own, which is what `_drain` is for.
+
     Marked delivered and committed *before* the request goes out, so this is
     at-most-once. A crash in between loses the turn, which is recoverable --
     the words are still in the transcript with Send now beside them. The other
@@ -2132,6 +2361,9 @@ def _inject(generation: Generation, chat_id: str, vision: bool) -> dict | None:
     """
     try:
         with session_scope() as db:
+            message = db.get(Message, generation.message_id)
+            if crowd_service.state_of(message) is not None:
+                return None
             waiting = _next_waiting(db, chat_id)
             if waiting is None:
                 return None
@@ -2265,7 +2497,13 @@ def _persist(generation: Generation, title: str, elapsed: float) -> None:
             # clears this when it is next opened. Not for a temporary chat:
             # there is no sidebar row for the dot, and the toast would name a
             # chat nobody can navigate to.
-            if generation.followers == 0 and not chat.temporary:
+            # 🚨 Once per *turn*, not once per speaker. `announce_later` has no
+            # dedupe of its own -- its docstring says so, because every site that
+            # calls it runs once per arrival -- so a five-model crowd with nobody
+            # watching would be nine web pushes and nine sidebar toasts for one
+            # question. The closing speaker is the arrival; everybody before it is
+            # the middle of one.
+            if generation.followers == 0 and not chat.temporary and _ends_the_turn(message):
                 chat.unread = True
                 chat.unread_notified = False
                 # And out to any browser that asked to be told, which is the

src/lembas/services/harness.py

@@ -166,6 +166,7 @@ def context_variables(
     user: User | None,
     tools: list[dict[str, Any]] | None,
     chat=None,
+    speaker=None,
 ) -> dict[str, str]:
     """What every ``{{name}}`` in a fragment resolves to for this request.
 
@@ -294,8 +295,12 @@ def context_variables(
         from lembas.services import chat as chat_service
         from lembas.services.subagent import ROLE_FRIEND
 
-        model = chat_service.model_for(db, chat)
+        # The *answering* model, not the chat's: telling a crowd member it is the
-        values["model_name"] = model.label if model is not None else chat.model_id
+        # main model is a lie it will then reason from, and its personality is
+        # keyed on whichever model is speaking.
+        speaking = speaker or chat_service.speaker_for(db, chat)
+        model = chat_service.model_row(db, speaking)
+        values["model_name"] = model.label if model is not None else speaking.model_id
         # Naming the bases a chat is scoped to matters: without it the model
         # cannot tell "there is nothing about this" from "I am only allowed to
         # see the contracts folder", and phrases a miss as the former.
@@ -333,7 +338,7 @@ def context_variables(
         # reason the roster and the tool are one checkbox rather than two.
         if "friend" in families:
             values["model_roster"] = chat_service.roster_block(
-                db, user, exclude=chat.model_id
+                db, user, exclude=speaking.model_id
             )
 
         if "persona" in families:
@@ -341,7 +346,7 @@ def context_variables(
             # administrator's default until the model has written one with them;
             # and this model's impression of them, which has no default and never
             # could.
-            key = chat.model_id
+            key = speaking.model_id
             values["persona"] = personas_service.block(db, key, user)
             values["person_view"] = personas_service.view_block(db, key, user)
 
@@ -512,12 +517,13 @@ def compose(
     user: User | None,
     tools: list[dict[str, Any]] | None,
     chat=None,
+    speaker=None,
 ) -> str:
     """The operational preamble for this request, or "" when there is nothing to say."""
     offered = tools or []
     return compose_from(
         db,
-        variables=context_variables(db, user, offered, chat),
+        variables=context_variables(db, user, offered, chat, speaker),
         families=_families(db, offered),
         has_tools=bool(offered),
     )

src/lembas/services/prompts.py

@@ -182,6 +182,14 @@ VARIABLES: tuple[Variable, ...] = (
         "Per model and per person, so two models may hold different views and "
         "nobody sees anybody else's. The person can read and delete it.",
     ),
+    Variable(
+        "crowd_speaker",
+        "The model being quoted",
+        "Inside the crowd fragments only: the name of the model whose words "
+        "follow, or whose turn it is. Blank everywhere else, because it is a "
+        "property of one quotation rather than of a request — which is why the "
+        "legend cannot show you a value for it.",
+    ),
     Variable(
         "timezone",
         "Timezone",
@@ -1995,6 +2003,106 @@ BUILTIN: tuple[Fragment, ...] = (
             "{{transcript}}"
         ),
     ),
+    Fragment(
+        key="crowd.said",
+        label="Quoting another model in a crowd",
+        group=GROUP_TASKS,
+        order=450,
+        variables=("crowd_speaker",),
+        hint="What another speaker's answer is labelled as when it reaches this "
+        "one. It matters more than it looks: sent unlabelled, every earlier reply "
+        "arrives as something *this* model said, so it defends sentences it never "
+        "wrote and cannot disagree with them — which is the whole point of the "
+        "way back. Relabelling is also what keeps the history alternating, which "
+        "several chat templates require.",
+        default="{{crowd_speaker}} answered:",
+    ),
+    Fragment(
+        key="crowd.turn",
+        label="A crowd member's turn on the way out",
+        group=GROUP_TASKS,
+        order=451,
+        hint="Added as the last turn when a member speaks on the forward pass. "
+        "The failure to word against is a member that repeats what has already "
+        "been said in different words, which is what makes a crowd feel like an "
+        "echo rather than a second opinion.",
+        default=(
+            "You are one of several models answering this. The answers above are "
+            "quoted with the name of whoever wrote them; yours comes next.\n"
+            "\n"
+            "Add what is missing, correct what is wrong, and say what you would "
+            "have done differently. Do not restate what has already been said to "
+            "show that you agree with it — if you have nothing to add, say so in "
+            "one line and stop. Be brief: somebody is reading all of these."
+        ),
+    ),
+    Fragment(
+        key="crowd.disagree",
+        label="A crowd member's turn on the way back",
+        group=GROUP_TASKS,
+        order=452,
+        hint="Added as the last turn on the backward pass, which is where the "
+        "value of a crowd actually is: everybody has now been heard, and this is "
+        "the chance to object. Worded to ask for disagreement rather than for a "
+        "summary, because a model asked to review will produce a review whether "
+        "it has one or not.",
+        default=(
+            "Everybody has now answered. Read the whole exchange again.\n"
+            "\n"
+            "Do you disagree with anything said above — a claim that is wrong, a "
+            "risk nobody named, an answer to the wrong question? Say so plainly, "
+            "and say which part you mean. **If you have no disagreement, reply "
+            "with one short sentence saying so and nothing else.** Do not "
+            "summarise, do not praise the other answers, and do not repeat your "
+            "own."
+        ),
+    ),
+    Fragment(
+        key="crowd.close",
+        label="The main model's last word, with another round available",
+        group=GROUP_TASKS,
+        order=453,
+        hint="The main model's closing turn when it can still ask for another "
+        "round. Its own fragment rather than a sentence inside the one below, "
+        "because inviting a choice a model cannot express is worse than not "
+        "offering it: on a model without the tools capability there is no "
+        "crowd_again to call, and that is the case the next fragment covers.",
+        default=(
+            "You opened this and you are closing it. The others have answered and "
+            "have had the chance to disagree.\n"
+            "\n"
+            "Write the answer the person actually asked for. Take what the others "
+            "got right, say where you disagree with them and why, and name "
+            "anything still unresolved rather than papering over it. Attribute "
+            "what you took from whom.\n"
+            "\n"
+            "If the disagreement is real and another round would settle it, call "
+            "crowd_again and say what you want them to address. Do not call it "
+            "because the discussion was interesting — every round costs the person "
+            "another wait."
+        ),
+    ),
+    Fragment(
+        key="crowd.close_final",
+        label="The main model's last word, with no round left",
+        group=GROUP_TASKS,
+        order=454,
+        hint="The same turn when another round is not on offer — the round limit "
+        "is reached, or this model has no tools and so cannot ask. It says the "
+        "answer has to be final rather than inviting a choice that would be "
+        "ignored, which is the difference between a feature and a feature that "
+        "looks like one.",
+        default=(
+            "You opened this and you are closing it, and this is the last turn: "
+            "there will be no further round.\n"
+            "\n"
+            "Write the answer the person actually asked for. Take what the others "
+            "got right, say where you disagree with them and why, and attribute "
+            "what you took from whom. Where the disagreement is unresolved, say so "
+            "and say what would settle it — that is more useful than a confident "
+            "answer papered over the top of it."
+        ),
+    ),
     Fragment(
         key="task.compact_lead",
         label="How a summary is introduced",

src/lembas/services/settings_store.py

@@ -32,6 +32,7 @@ AGENTS = "agents"
 IMAGES = "images"
 SCHEDULES = "schedules"
 SUBAGENTS = "subagents"
+CROWD = "crowd"
 BRANDING = "branding"
 EXTRACTION = "extraction"
 
@@ -343,6 +344,41 @@ def _schedules_defaults() -> dict[str, Any]:
     }
 
 
+def _crowd_defaults() -> dict[str, Any]:
+    """Several models answering one turn, in order, then again in reverse.
+
+    Off until an administrator turns it on, and the reason is arithmetic: one
+    turn costs **models x rounds x 2 - 1** replies, so four models over two
+    rounds is fifteen. On a single local endpoint every change of speaker is also
+    a model load, because llama-swap holds one at a time.
+
+    The owner's own warning, recorded because it is the failure this feature
+    actually has: *larger crowds of smaller models -- and sometimes of bigger
+    ones -- start cycling, or never stop.* So the numbers below are a ceiling
+    reached by ordinary work, not a runaway backstop, which is the opposite of
+    how `subagents.max_rounds` is set and is deliberate: a round of a crowd is a
+    visible, expensive thing somebody is waiting through.
+    """
+    return {
+        "enabled": False,
+        # Besides the chat's own model. Four speakers is already eight replies a
+        # turn at one round each.
+        "max_models": 4,
+        # One round is out-and-back: everyone answers, then everyone is asked
+        # whether they disagree, ending at the main model. Two is one chance to
+        # change its mind after hearing the objections, which is the whole point;
+        # three is where cycling starts.
+        "max_rounds": 2,
+        # The whole turn, across every speaker, so a member whose endpoint has
+        # stalled cannot hold a round open all afternoon.
+        "wall_seconds": 900,
+        # Whether a short "I agree" on the way back is collapsed in the
+        # transcript. On by default: N-1 bubbles saying nothing is what makes
+        # somebody switch the feature off, and the disagreements are the point.
+        "collapse_agreement": True,
+    }
+
+
 def _subagents_defaults() -> dict[str, Any]:
     """Delegating a piece of a reply to a second, unattended model.
 
@@ -389,6 +425,7 @@ _DEFAULTS: dict[str, Any] = {
     IMAGES: _images_defaults,
     SCHEDULES: _schedules_defaults,
     SUBAGENTS: _subagents_defaults,
+    CROWD: _crowd_defaults,
     # Whose instance this is. The defaults live in `services/branding.py`
     # beside the code that reads them, because every one of them is paired with
     # a label and a hint for the admin page and splitting the three across two
@@ -668,6 +705,22 @@ def subagents(db: DBSession) -> dict[str, Any]:
     return values
 
 
+def crowd(db: DBSession) -> dict[str, Any]:
+    """Crowd settings, clamped on read for the reason `agents` gives.
+
+    Every bound has a floor of one: a `max_models` of zero is the feature
+    switched off wearing the switch's clothes, and that is a thing to answer in
+    one place rather than two.
+    """
+    values = get_group(db, CROWD)
+    values["max_models"] = min(max(int(values.get("max_models") or 1), 1), 8)
+    values["max_rounds"] = min(max(int(values.get("max_rounds") or 1), 1), 5)
+    values["wall_seconds"] = min(max(int(values.get("wall_seconds") or 1), 60), 7200)
+    values["enabled"] = bool(values.get("enabled"))
+    values["collapse_agreement"] = bool(values.get("collapse_agreement"))
+    return values
+
+
 def images_ready(db: DBSession) -> bool:
     """Whether image generation can actually happen.
 

src/lembas/services/tool_labels.py

@@ -76,6 +76,8 @@ LABELS: dict[str, str] = {
     "subagent_run": "Helper",
     # A question put to one of the other models here.
     "ask_friend": "Asked another model",
+    # The main model sending a crowd round again.
+    "crowd_again": "Another round",
     # What a model keeps about itself and about the person it is talking to.
     "persona_write": "Personality rewritten",
     "impression_write": "Impression updated",
@@ -121,6 +123,7 @@ ICONS: dict[str, str] = {
     "schedule_cancel": "stop-circle",
     "subagent_run": "sparkle",
     "ask_friend": "users",
+    "crowd_again": "refresh",
     "persona_write": "user",
     "impression_write": "user",
     "memory_add": "star",
@@ -169,6 +172,7 @@ ACTIONS: dict[str, str] = {
     "schedule_cancel": "Stop a schedule",
     "subagent_run": "Send a helper",
     "ask_friend": "Ask another model",
+    "crowd_again": "Send the crowd round again",
     "persona_write": "Rewrite its own personality",
     "impression_write": "Update what it makes of you",
     "memory_add": "Remember something",
@@ -216,6 +220,9 @@ DETAIL_KEYS: dict[str, str] = {
     # question carrying a wrong assumption comes back as a confident answer
     # about the wrong thing -- the same reason `subagent_run` names the task.
     "ask_friend": "question",
+    # What the next round is for. The only field it has, and the one thing worth
+    # correcting before several models spend a reply each on it.
+    "crowd_again": "focus",
     # The whole text, because for these two the text *is* the thing being agreed
     # to: there is no shorter field that says what the model would become.
     "persona_write": "content",

src/lembas/services/tools.py

@@ -162,6 +162,15 @@ FAMILY_FRIEND = "friend"
 # a model up: either it may form and keep opinions of this kind or it may not.
 FAMILY_PERSONA = "persona"
 
+# Sending a crowd round again. Its own family so `harness._families` can map the
+# name back to one, and deliberately **not in `FAMILIES`**: that tuple is the list
+# of things an administrator switches on, and this is mechanism. Being in it would
+# mint a `tool_crowd` capability checkbox and demand a `tools.crowd` permission
+# that does not exist -- which, because `_family_allowed` falls through to
+# `allowed.get(...)`, would mean the tool could never be offered at all. Its real
+# gate is `resolve_tools(crowd_again=…)`: one turn of one round.
+FAMILY_CROWD = "crowd"
+
 # The built-in families, in the order they are offered.
 FAMILIES = (
     FAMILY_SEARCH,
@@ -1635,6 +1644,13 @@ def _family_allowed(
         # separate switch would be a second door to the cost with nothing
         # naming it. `Helpers` on /admin/agents is where both are bounded.
         return bool(allowed.get("tools.friend") and subagents)
+    if gate == FAMILY_CROWD:
+        # Always allowed, because whether it is *offered* is decided before this:
+        # `resolve_tools` puts it in the book only on the main model's closing turn
+        # with a round still left. A permission here would be a second switch for
+        # one already-enabled feature, and an absent one would silently make the
+        # crowd a single round for ever.
+        return True
     if gate in (
         FAMILY_CUSTOM,
         FAMILY_MCP,
@@ -1721,6 +1737,13 @@ def _friend_defs() -> list[ToolDef]:
     return subagent_service.friend_tool_defs()
 
 
+def _crowd_defs() -> list[ToolDef]:
+    """The go-round-again tool. Imported inside the call for the reason above."""
+    from lembas.services import crowd as crowd_service
+
+    return crowd_service.tool_defs()
+
+
 def _image_defs(db: DBSession, values: dict | None = None) -> list[ToolDef]:
     """The image tool, whose schema carries this instance's own choices.
 
@@ -1776,6 +1799,7 @@ def registry(db: DBSession) -> dict[str, ToolDef]:
             *_schedule_defs(),
             *_subagent_defs(),
             *_friend_defs(),
+            *_crowd_defs(),
         ]
     )
 
@@ -1786,13 +1810,31 @@ def families(db: DBSession) -> tuple[str, ...]:
     return (*FAMILIES, *rows)
 
 
-def resolve_tools(db: DBSession, chat: Chat, user: User | None) -> ToolSet:
+def resolve_tools(
-    """Every tool this chat may call right now, with its runner attached."""
+    db: DBSession,
+    chat: Chat,
+    user: User | None,
+    speaker=None,
+    *,
+    crowd_turn=None,
+    crowd_again: bool = False,
+) -> ToolSet:
+    """Every tool this chat may call right now, with its runner attached.
+
+    The capabilities are the **answering** model's. `tools` being off is the first
+    gate and returns nothing at all, so handing a crowd member the main model's
+    switches would offer a tool list to an endpoint that rejects the request for
+    carrying one.
+    """
     from lembas.security import permissions
     from lembas.services import chat as chat_service
 
     capabilities = {}
-    model = chat_service.model_for(db, chat)
+    model = (
+        chat_service.model_row(db, speaker)
+        if speaker is not None
+        else chat_service.model_for(db, chat)
+    )
     if model is not None:
         capabilities = model.capabilities_json or {}
 
@@ -1819,6 +1861,12 @@ def resolve_tools(db: DBSession, chat: Chat, user: User | None) -> ToolSet:
             *(_schedule_defs() if schedules_on else []),
             *(_subagent_defs() if subagents_on else []),
             *(_friend_defs() if subagents_on else []),
+            # Only on the closing turn, and only with a round left. Not gated on a
+            # capability or a permission: a tool that exists on exactly one turn of
+            # one feature is mechanism, and an administrator switching it off would
+            # be switching off the main model's ability to use the feature it
+            # already enabled.
+            *(_crowd_defs() if crowd_again else []),
         ]
     )
 
@@ -1832,6 +1880,26 @@ def resolve_tools(db: DBSession, chat: Chat, user: User | None) -> ToolSet:
     off = scoped_off(chat)
     empty_library = not skills_service.count_enabled(db, user, exclude=scoped_skills_off(chat))
 
+    # What a crowd speaker may do, which is narrower than what the chat may.
+    if crowd_turn is not None:
+        from lembas.services import crowd as crowd_service
+
+        if crowd_turn.phase == crowd_service.PHASE_BACK:
+            # The way back is "do you disagree with any of this", which needs
+            # nothing looked up: everything it is about is already in front of it.
+            # An empty toolset also guarantees the turn ends in words, which is the
+            # shape `_wrap_up` relies on.
+            return ToolSet()
+        if not crowd_turn.is_main:
+            # A member answers a machine-composed instruction with several models'
+            # words quoted into it, and nobody is waiting on *it* in particular.
+            # So: it cannot stop the round for an approval or a question -- one
+            # card would park every remaining speaker for `approval_timeout` -- it
+            # cannot fan out, and it cannot rewrite a personality under wording it
+            # did not choose. The same set `unattended` withdraws, for the same
+            # reasons, applied for a different one.
+            off = off | {FAMILY_ASK, FAMILY_SUBAGENT, FAMILY_FRIEND, FAMILY_PERSONA}
+
     # A scheduled task runs with nobody present, so `ask_user` cannot work here:
     # it pauses the reply and waits for a POST that will never come, until
     # `approval_timeout` expires -- a run that silently does nothing for fifteen
@@ -2017,10 +2085,21 @@ def context_for(
     chat: Chat | None = None,
     *,
     tools: ToolSet | None = None,
+    speaker=None,
 ) -> ToolContext:
-    """The snapshot a running tool needs, taken while the session is open."""
+    """The snapshot a running tool needs, taken while the session is open.
+
+    `speaker` is the model answering, and it decides which model a tool acts *as*:
+    which personality `persona_write` rewrites, and whose endpoint the image
+    reviewer and the Preserve-VRAM unload reach for. It defaults to the chat's own
+    model.
+    """
+    from lembas.services import chat as chat_service
     from lembas.services.agent import session as agent_session
 
+    if chat is not None and speaker is None:
+        speaker = chat_service.speaker_for(db, chat)
+
     return ToolContext(
         agent=agent_session.resolve(db, chat, user) if chat is not None else None,
         owner_id=user.id if user else "",
@@ -2029,8 +2108,8 @@ def context_for(
         image_config=settings_store.images(db),
         image_workflow_id=(chat.image_workflow_id or "") if chat is not None else "",
         image_checkpoint=(chat.image_checkpoint or "") if chat is not None else "",
-        model_id=(chat.model_id or "") if chat is not None else "",
+        model_id=(speaker.model_id or "") if speaker is not None else "",
-        connection_id=(chat.connection_id or "") if chat is not None else "",
+        connection_id=(speaker.connection_id or "") if speaker is not None else "",
         base_ids=[base.id for base in chat.knowledge_bases] if chat is not None else [],
         skills_off=scoped_skills_off(chat),
         tools=tools.by_name if tools is not None else None,

src/lembas/web/static/css/app.css

@@ -417,6 +417,12 @@ input.visually-hidden[type="checkbox"] {
   color: var(--ink-muted);
 }
 .badge--leaf { background: var(--leaf-soft); color: var(--leaf); }
+/* A crowd's backward pass: everybody has answered and each is being asked
+   whether it disagrees. Quieter than an answer, because most of these are one
+   line saying "no" -- and deliberately *not* hidden, because the one that says
+   yes is the whole reason the feature exists. */
+.msg--crowd-back .msg__body { color: var(--ink-muted); }
+.msg--crowd-back .msg__author { font-weight: 500; }
 .badge--success { background: var(--success-soft); color: var(--success); }
 .badge--danger { background: var(--danger-soft); color: var(--danger); }
 .badge--warning { background: var(--warning-soft); color: var(--warning); }

src/lembas/web/templates/admin/agents.html

@@ -445,6 +445,93 @@
   button was pressed, which is what keeps each group's save handler writing one
   key.
 #}
+{# A third settings group on this page, saved by its own form -- the reason the
+   Helpers card gives. A crowd is not an agent-chat feature either, but this is the
+   page somebody opens to find out what one turn may set going. #}
+<form method="post" action="/admin/agents/crowd" class="form-grid">
+  <section class="card">
+    <h2 class="card__title">A crowd</h2>
+    <p class="field__hint">
+      A chat can have more than one model in it. The chat's own model answers, then
+      each of the others in turn; then the order runs <strong>backwards</strong>,
+      each one asked whether it disagrees with anything; and it ends back at the
+      first, which either closes or sends them round again.
+    </p>
+
+    <div class="alert">
+      {{ icon("warning", "icon--sm") }}
+      <span>
+        One turn costs <strong>models × rounds × 2 − 1</strong> replies — four
+        models over two rounds is fifteen — and on a single local endpoint every
+        change of speaker also loads a different model. Larger crowds of smaller
+        models, and sometimes of bigger ones, start going round in circles: that is
+        what the round limit is for, and it is a limit ordinary work will reach
+        rather than a runaway backstop.
+      </span>
+    </div>
+
+    <div class="field">
+      <label class="checkbox">
+        <input type="checkbox" name="enabled" value="true"
+               {{ 'checked' if crowd.enabled }}>
+        <span>Let a chat have a crowd</span>
+      </label>
+      <p class="field__hint">
+        Off by default. With it on, each chat's settings panel offers the other
+        models; a chat with none ticked behaves exactly as it always has.
+      </p>
+    </div>
+
+    <div class="field">
+      <label class="field__label" for="crowd_max_models">Most models besides the chat's own</label>
+      <input class="input" id="crowd_max_models" name="max_models"
+             type="number" min="1" max="8" step="1" value="{{ crowd.max_models }}">
+      <p class="field__hint">
+        Four is already eight replies a turn at one round each. More voices past
+        that tend to repeat each other rather than add anything.
+      </p>
+    </div>
+
+    <div class="field">
+      <label class="field__label" for="crowd_max_rounds">Most rounds</label>
+      <input class="input" id="crowd_max_rounds" name="max_rounds"
+             type="number" min="1" max="5" step="1" value="{{ crowd.max_rounds }}">
+      <p class="field__hint">
+        A round is out and back. Two gives the first model one chance to change its
+        mind after hearing the objections, which is the point of the whole thing;
+        three is where going in circles starts.
+      </p>
+    </div>
+
+    <div class="field">
+      <label class="field__label" for="crowd_wall_seconds">Longest a turn may take</label>
+      <input class="input" id="crowd_wall_seconds" name="wall_seconds"
+             type="number" min="60" max="7200" step="30" value="{{ crowd.wall_seconds }}">
+      <p class="field__hint">
+        Across every speaker, not each. A member whose endpoint has stalled cannot
+        then hold the round open all afternoon.
+      </p>
+    </div>
+
+    <div class="field">
+      <label class="checkbox">
+        <input type="checkbox" name="collapse_agreement" value="true"
+               {{ 'checked' if crowd.collapse_agreement }}>
+        <span>Fold away a short "I agree" on the way back</span>
+      </label>
+      <p class="field__hint">
+        The disagreements are what a crowd is for; a column of bubbles saying
+        nothing is what makes somebody switch it off. The text is still there
+        behind a disclosure.
+      </p>
+    </div>
+
+    <div class="btn-row">
+      <button class="btn btn--primary" type="submit">Save</button>
+    </div>
+  </section>
+</form>
+
 <form method="post" action="/admin/agents/subagents" class="form-grid">
   <section class="card">
     <h2 class="card__title">Helpers</h2>

src/lembas/web/templates/chat/_message.html

@@ -29,8 +29,17 @@
   `msg--machine` only overrides what should differ.
 #}
 {% set machine = (message.role == "user" and message.machine) %}
+{#
+  Where this bubble sits in a crowd round, if it is in one. Nine bubbles for one
+  question need orienting, and a `<details>` wrapper round the round is the wrong
+  way to do it: bubbles arrive with `beforeend:#thread`, which appends *after* any
+  container, so the live and reloaded renderings would disagree and a reload would
+  rearrange nine bubbles under the reader. A chip on each one is the same markup
+  either way.
+#}
+{% set crowd = message.crowd_json or None %}
 
-<article class="msg msg--{{ message.role }}{{ ' msg--machine' if machine }}{{ ' msg--queued' if queued }}"
+<article class="msg msg--{{ message.role }}{{ ' msg--machine' if machine }}{{ ' msg--queued' if queued }}{{ ' msg--crowd-back' if crowd and crowd.get('phase') == 'back' }}"
          id="msg-{{ message.id }}"
   {% if streaming %}
          hx-ext="sse"
@@ -74,6 +83,33 @@
       {# Only worth showing when it adds something the author line does not. #}
       <span class="msg__model" title="{{ message.model_id }}">{{ message.model_id }}</span>
       {% endif %}
+      {% if crowd %}
+      {# Which speaker, which pass. The count is of speakers rather than of
+         replies: a round produces more bubbles than it has models in it. #}
+      <span class="badge">
+        {% if crowd.get("phase") == "out" %}
+          {{ crowd.get("index", 0) + 1 }} of {{ crowd.get("of", 1) }}
+        {% elif crowd.get("phase") == "back" %}
+          on the way back
+        {% else %}
+          closing
+        {% endif %}
+        {% if crowd.get("round", 1) > 1 %} · round {{ crowd.get("round") }}{% endif %}
+      </span>
+      {% if crowd.get("stopped") %}
+      {# Why a round ended, where it ended. Without this a crowd that ran out of
+         rounds or time simply stops, which reads as the feature failing. #}
+      <span class="badge badge--warning" title="The round ended here">
+        {% if crowd.get("stopped") == "rounds" %}
+          no rounds left
+        {% elif crowd.get("stopped") == "time" %}
+          out of time
+        {% else %}
+          two endpoints failed
+        {% endif %}
+      </span>
+      {% endif %}
+      {% endif %}
     </header>
 
     {% if message.attachments %}

src/lembas/web/templates/chat/index.html

@@ -243,6 +243,52 @@
         </div>
         {% endif %}
 
+        {% if crowd_available %}
+        {# Who else answers. Nothing ticked is every chat that has ever existed:
+           one model, answering on its own. #}
+        <div class="field">
+          <label class="field__label">Crowd</label>
+          <form hx-patch="/api/chats/{{ chat.id }}" hx-swap="none" hx-trigger="change">
+            {# Always submitted, for the reason the bases above are. #}
+            <input type="hidden" name="crowd_model_ids" value="">
+            <div class="checkbox-row">
+              {% for model in crowd_available %}
+              <label class="checkbox">
+                <input type="checkbox" name="crowd_model_ids" value="{{ model.model_id }}"
+                       {{ 'checked' if model.model_id in crowd_member_ids }}>
+                <span>{{ model.label }}</span>
+              </label>
+              {% endfor %}
+            </div>
+          </form>
+          <p class="field__hint">
+            {% if crowd_member_ids %}
+              {{ crowd_member_ids|length + 1 }} models answer each turn: this one
+              first, then the others, then back through them asking whether they
+              disagree, ending here.
+              <strong>That is {{ crowd_replies }} replies a turn</strong>, and up to
+              {{ crowd_rounds }} rounds of it.
+            {% else %}
+              Tick a model to have it answer after this one, then be asked whether
+              it disagrees. Useful for a second opinion; expensive, because each
+              one is a whole reply, and slow on one local endpoint because every
+              change of speaker loads a different model.
+            {% endif %}
+          </p>
+          {# Outside the two branches above, deliberately. A member whose model has
+             gone is filtered out of `crowd_member_ids`, so if it was the only one
+             this would fall into the "tick a model" branch and never mention the
+             row that is still there -- which is the one thing somebody needs to
+             know to tidy it up. #}
+          {% if crowd_skipped %}
+          <p class="field__hint">
+            Skipped, because you cannot reach {{ "them" if crowd_skipped|length > 1 else "it" }}
+            any more: <s>{{ crowd_skipped|join(", ") }}</s>. Untick to clear.
+          </p>
+          {% endif %}
+        </div>
+        {% endif %}
+
         {% if can.get("chat.params") %}
         <div class="grid grid--3">
           <div class="field">

tests/test_agent_interaction.py

@@ -232,7 +232,9 @@ async def test_an_unanswered_question_expires_and_the_reply_finishes(db, user_id
     assert generation.tool_events[0]["status"] == "error"
 
 
-def _fast_context(db, user, chat=None, *, tools=None):
+def _fast_context(db, user, chat=None, *, tools=None, **rest):
+    """`**rest` so a new keyword on the real `context_for` does not fail this as
+    an IndexError three assertions later. It grew `speaker` in 1.6.0."""
     from lembas.services import tools as tools_service
 
     context = tools_service.ToolContext(

tests/test_chat.py

@@ -637,6 +637,15 @@ def test_editing_rewinds_and_discards_later_messages(
     first_user = db.scalars(
         select(Message).where(Message.role == "user").order_by(Message.created_at)
     ).first()
+    # The ids as strings, taken now: these rows are about to be deleted, and an
+    # ORM instance read afterwards raises ObjectDeletedError.
+    discarded_ids = set(
+        db.scalars(
+            select(Message.id).where(
+                Message.chat_id == chat_id, Message.role == "assistant"
+            )
+        )
+    )
     client.post(
         f"/api/chats/{chat_id}/messages/{first_user.id}/edit",
         data={"content": "first, revised"},
@@ -648,8 +657,17 @@ def test_editing_rewinds_and_discards_later_messages(
     remaining = db.scalars(select(Message).order_by(Message.created_at)).all()
     assert [m.role for m in remaining] == ["user", "assistant"]
     assert remaining[0].content == "first, revised"
-    # The fresh assistant row is incomplete, which is what restarts the stream.
+    # A *fresh* assistant row, which is what restarts the stream: a different row
-    assert remaining[1].complete is False
+    # from the one that was discarded, with nothing written into it yet.
+    #
+    # ⚠ Deliberately not `complete is False`. A real generation is started here
+    # against the fixture's unreachable endpoint, and it does finish -- it errors
+    # with "could not reach" and `_persist` marks the row complete. Whether that
+    # has happened by the time this line runs is a race, and asserting on it made
+    # this test pass only while that failure stayed slower than the rest of the
+    # request. It began flaking the moment unrelated work shifted the timing.
+    assert remaining[1].id not in discarded_ids
+    assert remaining[1].content == ""
 
 
 def test_a_rewind_takes_a_message_written_in_the_same_microsecond(

tests/test_crowd_chain.py

@@ -0,0 +1,347 @@
+"""One user turn, several speakers, chained.
+
+`_advance_crowd` is the shell around the pure scheduler, so what is asserted here
+is the part the scheduler cannot see: which rows exist, when, and how many. The
+producer is replaced, so nothing here talks to an endpoint — what matters is the
+chat each speaker is handed and the invariant that holds between them.
+
+**Exactly one incomplete assistant row at every observation.** That is the whole
+reason this shape was chosen over one generation writing many bubbles: it is what
+`_reply_in_flight`, `_too_many_replies`, `wake.lock_for` and the superseded guards
+in `_persist`/`_drain` all already rely on, and its symptom when broken is a Stop
+button pointing at whichever bubble comes first in the document.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import select
+
+from lembas.db.models import (
+    ROLE_ASSISTANT,
+    ROLE_USER,
+    Chat,
+    Connection,
+    CrowdMember,
+    Message,
+    Model,
+    User,
+)
+from lembas.services import chat as chat_service
+from lembas.services import crowd as crowd_service
+from lembas.services import generation as generation_service
+from lembas.services import settings_store
+from lembas.services.crypto import encrypt
+
+MEMBERS = ("second-model", "third-model")
+
+
+@pytest.fixture(autouse=True)
+def empty_registry():
+    yield
+    generation_service._RUNNING.clear()
+    generation_service._TASKS.clear()
+
+
+@pytest.fixture(autouse=True)
+def crowd_on(db, registered):
+    settings_store.update(db, {"enabled": True}, key=settings_store.CROWD)
+    connection = Connection(
+        name="Test", base_url="http://127.0.0.1:1", api_key_encrypted=encrypt("")
+    )
+    db.add(connection)
+    db.commit()
+    for index, name in enumerate(("main-model", *MEMBERS)):
+        db.add(
+            Model(
+                connection_id=connection.id,
+                model_id=name,
+                display_name=name,
+                position=index,
+                capabilities_json={"tools": True},
+            )
+        )
+    db.commit()
+
+
+@pytest.fixture
+def started(monkeypatch):
+    """Every chat id `ensure` was asked to start a reply in, in order."""
+    calls: list[tuple[str, str]] = []
+
+    def _fake_ensure(chat_id, message_id):
+        calls.append((chat_id, message_id))
+
+    monkeypatch.setattr(generation_service, "ensure", _fake_ensure)
+    return calls
+
+
+def _user(db) -> User:
+    return db.scalars(select(User).order_by(User.created_at)).first()
+
+
+def _crowd_chat(db, members=MEMBERS) -> Chat:
+    connection = db.scalars(select(Connection)).first()
+    chat = Chat(
+        user_id=_user(db).id, title="t", model_id="main-model", connection_id=connection.id
+    )
+    db.add(chat)
+    db.commit()
+    for index, name in enumerate(members):
+        db.add(CrowdMember(chat_id=chat.id, model_id=name, position=index))
+    db.commit()
+    return chat
+
+
+def _opening_reply(db, chat) -> Message:
+    """The main model's first answer: a user turn and a finished assistant one."""
+    chat_service.create_message(db, chat, ROLE_USER, "What should we do?")
+    reply = chat_service.create_message(
+        db, chat, ROLE_ASSISTANT, "Rewrite it.", model_id=chat.model_id
+    )
+    return reply
+
+
+def _advance(db, chat, message, **kwargs) -> bool:
+    generation = generation_service.Generation(chat_id=chat.id, message_id=message.id)
+    for key, value in kwargs.items():
+        setattr(generation, key, value)
+    generation_service._RUNNING[message.id] = generation
+    try:
+        return generation_service._advance_crowd(generation)
+    finally:
+        generation_service._RUNNING.pop(message.id, None)
+
+
+def _incomplete(db, chat) -> list[Message]:
+    return list(
+        db.scalars(
+            select(Message).where(
+                Message.chat_id == chat.id, Message.complete.is_(False)
+            )
+        )
+    )
+
+
+def _run_round(db, chat, started, *, answers: int = 12) -> list[Message]:
+    """Walk a whole round by finishing each speaker as it is created."""
+    order: list[Message] = []
+    message = _opening_reply(db, chat)
+    for _ in range(answers):
+        assert len(_incomplete(db, chat)) == 0, "a row was left incomplete"
+        if not _advance(db, chat, message):
+            break
+        db.expire_all()
+        fresh = _incomplete(db, chat)
+        assert len(fresh) == 1, f"{len(fresh)} replies in flight at once"
+        message = fresh[0]
+        order.append(message)
+        message.content = "Something."
+        message.complete = True
+        db.commit()
+    return order
+
+
+# --- The chain ----------------------------------------------------------------
+def test_a_whole_round_speaks_in_order(db, started):
+    chat = _crowd_chat(db)
+    order = _run_round(db, chat, started)
+
+    assert [m.model_id for m in order] == [
+        "second-model",   # out
+        "third-model",    # out
+        "second-model",   # back
+        "main-model",     # close
+    ]
+
+
+def test_each_speaker_is_started_through_ensure(db, started):
+    chat = _crowd_chat(db)
+    order = _run_round(db, chat, started)
+    assert [message_id for _chat_id, message_id in started] == [m.id for m in order]
+
+
+def test_the_round_is_recorded_on_every_row(db, started):
+    chat = _crowd_chat(db)
+    order = _run_round(db, chat, started)
+
+    phases = [crowd_service.state_of(m).phase for m in order]
+    assert phases == [
+        crowd_service.PHASE_OUT,
+        crowd_service.PHASE_OUT,
+        crowd_service.PHASE_BACK,
+        crowd_service.PHASE_CLOSE,
+    ]
+    # And they all belong to the same question.
+    anchors = {crowd_service.state_of(m).turn for m in order}
+    assert len(anchors) == 1
+
+
+def test_each_speaker_carries_its_own_connection(db, started):
+    """So `speaker_for` resolves the pair rather than guessing at the id."""
+    chat = _crowd_chat(db)
+    order = _run_round(db, chat, started)
+    for message in order:
+        assert chat_service.speaker_for(db, chat, message).model_id == message.model_id
+
+
+def test_a_chat_with_no_crowd_is_not_chained(db, started):
+    chat = _crowd_chat(db, members=())
+    message = _opening_reply(db, chat)
+    assert _advance(db, chat, message) is False
+    assert started == []
+
+
+def test_the_feature_switch_holds_the_whole_thing(db, started):
+    settings_store.update(db, {"enabled": False}, key=settings_store.CROWD)
+    chat = _crowd_chat(db)
+    message = _opening_reply(db, chat)
+    assert _advance(db, chat, message) is False
+
+
+def test_the_member_cap_trims_the_crowd(db, started):
+    settings_store.update(db, {"max_models": 1}, key=settings_store.CROWD)
+    chat = _crowd_chat(db)
+    order = _run_round(db, chat, started)
+    # One member: out to it, then straight back to the main model.
+    assert [m.model_id for m in order] == ["second-model", "main-model"]
+
+
+def test_a_member_whose_model_has_gone_is_skipped(db, started):
+    """Membership is text with no foreign key, so a model that disappears upstream
+    leaves a row behind. Skipping it is the point -- a cascade would have deleted
+    the crowd out of every chat on the next Test & refresh."""
+    chat = _crowd_chat(db)
+    gone = db.scalar(select(Model).where(Model.model_id == "third-model"))
+    db.delete(gone)
+    db.commit()
+
+    order = _run_round(db, chat, started)
+
+    assert "third-model" not in [m.model_id for m in order]
+    assert [m.model_id for m in order] == ["second-model", "main-model"]
+    # The row is still there, so a screen can say it was skipped.
+    assert any(row.model_id == "third-model" for row in db.get(Chat, chat.id).crowd)
+
+
+def test_a_disabled_model_is_skipped_too(db, started):
+    chat = _crowd_chat(db)
+    off = db.scalar(select(Model).where(Model.model_id == "third-model"))
+    off.enabled = False
+    db.commit()
+    assert "third-model" not in [m.model_id for m in _run_round(db, chat, started)]
+
+
+# --- The refusals -------------------------------------------------------------
+def test_stop_ends_the_round(db, started):
+    """Not just the speaker writing at the time. `_drain` refuses after a stop for
+    the same reason: somebody asked for it to end."""
+    chat = _crowd_chat(db)
+    message = _opening_reply(db, chat)
+    assert _advance(db, chat, message, stopped=True) is False
+    assert started == []
+
+
+def test_a_superseded_generation_advances_nothing(db, started):
+    """The guard `_persist` and `_drain` both carry."""
+    chat = _crowd_chat(db)
+    message = _opening_reply(db, chat)
+    other = generation_service.Generation(chat_id=chat.id, message_id=message.id)
+    generation_service._RUNNING[message.id] = other
+    try:
+        mine = generation_service.Generation(chat_id=chat.id, message_id=message.id)
+        assert generation_service._advance_crowd(mine) is False
+    finally:
+        generation_service._RUNNING.pop(message.id, None)
+    assert started == []
+
+
+def test_regenerating_a_speaker_does_not_fork_the_round(db, started):
+    """`restart` re-runs `_run`, whose `finally` advances the crowd again -- and the
+    speakers after it already exist. Without the newest-message guard, regenerating
+    member 2 creates a second member 3 and two chains race down one turn."""
+    chat = _crowd_chat(db)
+    order = _run_round(db, chat, started)
+    before = len(list(db.scalars(select(Message).where(Message.chat_id == chat.id))))
+
+    # The second speaker is regenerated: it is no longer the newest row.
+    assert _advance(db, chat, order[0]) is False
+
+    db.expire_all()
+    after = len(list(db.scalars(select(Message).where(Message.chat_id == chat.id))))
+    assert after == before
+
+
+def test_one_speaker_failing_is_skipped(db, started):
+    chat = _crowd_chat(db)
+    message = _opening_reply(db, chat)
+    assert _advance(db, chat, message) is True
+    db.expire_all()
+    second = _incomplete(db, chat)[0]
+    second.complete = True
+    second.error = "the endpoint fell over"
+    db.commit()
+
+    assert _advance(db, chat, second, error="the endpoint fell over") is True
+    db.expire_all()
+    assert _incomplete(db, chat)[0].model_id == "third-model"
+
+
+def test_two_failures_in_a_row_abandon_the_round(db, started):
+    chat = _crowd_chat(db)
+    message = _opening_reply(db, chat)
+    _advance(db, chat, message)
+    db.expire_all()
+    second = _incomplete(db, chat)[0]
+    second.complete = True
+    db.commit()
+    _advance(db, chat, second, error="down")
+    db.expire_all()
+    third = _incomplete(db, chat)[0]
+    third.complete = True
+    db.commit()
+
+    assert _advance(db, chat, third, error="down") is False
+
+    db.expire_all()
+    assert crowd_service.state_of(db.get(Message, third.id)).stopped == (
+        crowd_service.STOPPED_ERRORS
+    )
+
+
+def test_why_a_round_stopped_is_written_where_it_stopped(db, started):
+    """So the transcript can say a round ended rather than simply ending."""
+    settings_store.update(db, {"max_rounds": 1}, key=settings_store.CROWD)
+    chat = _crowd_chat(db)
+    order = _run_round(db, chat, started)
+    closing = order[-1]
+
+    assert _advance(db, chat, closing, crowd_again=True) is False
+
+    db.expire_all()
+    assert crowd_service.state_of(db.get(Message, closing.id)).stopped == (
+        crowd_service.STOPPED_ROUNDS
+    )
+
+
+# --- Going round again --------------------------------------------------------
+def test_the_main_model_can_send_them_round_again(db, started):
+    settings_store.update(db, {"max_rounds": 2}, key=settings_store.CROWD)
+    chat = _crowd_chat(db)
+    order = _run_round(db, chat, started)
+    closing = order[-1]
+
+    assert _advance(db, chat, closing, crowd_again=True) is True
+
+    db.expire_all()
+    fresh = _incomplete(db, chat)[0]
+    state = crowd_service.state_of(fresh)
+    assert state.round == 2
+    assert state.phase == crowd_service.PHASE_OUT
+    assert fresh.model_id == "second-model"
+
+
+def test_without_asking_the_round_is_over(db, started):
+    chat = _crowd_chat(db)
+    order = _run_round(db, chat, started)
+    assert _advance(db, chat, order[-1], crowd_again=False) is False

tests/test_crowd_payload.py

@@ -0,0 +1,252 @@
+"""What one crowd speaker is actually sent.
+
+Pure: `build_request` with a speaker, no generation and no endpoint. Two
+properties matter more than anything else here, and both are silent when wrong.
+
+**Another speaker's reply must not arrive as this one's own turn.** Sent verbatim,
+every assistant message in the payload reads as something *this* model wrote — so
+it defends sentences it never said and cannot disagree with them, which is the
+entire purpose of the backward pass.
+
+**The history has to alternate.** Several chat templates reject one that does not,
+and this project already works around it once: `task.compact_ack` exists so a
+compacted history still alternates. A crowd produces consecutive assistant turns
+by construction, so relabelling is what keeps it sendable at all.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import select
+
+from lembas.db.models import ROLE_ASSISTANT, ROLE_USER, Chat, Connection, Model, User
+from lembas.services import chat as chat_service
+from lembas.services import crowd as crowd_service
+from lembas.services import prompts as prompts_service
+from lembas.services.crypto import encrypt
+
+MODELS = ("main-model", "second-model", "third-model")
+
+
+@pytest.fixture(autouse=True)
+def three_models(db, registered):
+    connection = Connection(
+        name="Test", base_url="http://127.0.0.1:1", api_key_encrypted=encrypt("")
+    )
+    db.add(connection)
+    db.commit()
+    for index, name in enumerate(MODELS):
+        db.add(
+            Model(
+                connection_id=connection.id,
+                model_id=name,
+                display_name=name.replace("-model", "").title(),
+                position=index,
+                capabilities_json={"tools": True},
+            )
+        )
+    db.commit()
+
+
+def _user(db) -> User:
+    return db.scalars(select(User).order_by(User.created_at)).first()
+
+
+def _chat(db) -> Chat:
+    connection = db.scalars(select(Connection)).first()
+    chat = Chat(
+        user_id=_user(db).id, title="t", model_id="main-model", connection_id=connection.id
+    )
+    db.add(chat)
+    db.commit()
+    return chat
+
+
+def _round(db, chat, answers: list[tuple[str, str]]):
+    """A user turn, then one assistant reply per (model, text)."""
+    chat_service.create_message(db, chat, ROLE_USER, "What should we do?")
+    for model_id, text in answers:
+        chat_service.create_message(db, chat, ROLE_ASSISTANT, text, model_id=model_id)
+
+
+def _payload(db, chat, speaker_id: str, *, turn=None, again=False):
+    placeholder = chat_service.create_message(
+        db, chat, ROLE_ASSISTANT, "", complete_=False, model_id=speaker_id
+    )
+    if turn is not None:
+        placeholder.crowd_json = turn.as_json()
+        db.commit()
+    body = chat_service.build_request(
+        db, chat, upto=placeholder, user=_user(db), crowd_again=again
+    )
+    return body["messages"]
+
+
+def _roles(messages) -> list[str]:
+    return [m["role"] for m in messages if m["role"] != "system"]
+
+
+# --- Whose words are whose ----------------------------------------------------
+def test_another_speakers_answer_arrives_quoted_and_attributed(db):
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "Rewrite it in Rust.")])
+
+    messages = _payload(db, chat, "second-model")
+
+    quoted = [m for m in messages if "Rewrite it in Rust." in str(m["content"])]
+    assert quoted, "the other speaker's answer never reached this one"
+    assert quoted[0]["role"] == ROLE_USER, "it arrived as this model's own words"
+    assert "Main" in quoted[0]["content"], "it arrived unattributed"
+
+
+def test_a_speakers_own_earlier_turn_stays_its_own(db):
+    """Relabelling everything would be the same bug from the other side: a model
+    told that its own answer was somebody else's cannot be held to it."""
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "Mine."), ("second-model", "Theirs.")])
+
+    messages = _payload(db, chat, "main-model")
+
+    mine = [m for m in messages if "Mine." in str(m["content"])]
+    assert mine[0]["role"] == ROLE_ASSISTANT
+    theirs = [m for m in messages if "Theirs." in str(m["content"])]
+    assert theirs[0]["role"] == ROLE_USER
+
+
+def test_an_ordinary_one_model_chat_is_untouched(db):
+    """A chat with no other speaker in it must build the payload it always did.
+
+    Asserted as the property rather than by comparing two calls: `build_request`
+    resolves the harness and a bare `build_messages` does not, so comparing the two
+    would fail for a reason that has nothing to do with crowds -- which is what the
+    first version of this test did.
+    """
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "Just me.")])
+
+    messages = _payload(db, chat, "main-model")
+
+    assert _roles(messages) == [ROLE_USER, ROLE_ASSISTANT]
+    assert all("answered:" not in str(m["content"]) for m in messages)
+    # And nothing was appended: no crowd state on the row means no instruction.
+    assert messages[-1]["content"] == "Just me."
+
+
+# --- Alternation --------------------------------------------------------------
+@pytest.mark.parametrize("speaker_id", MODELS)
+def test_no_two_turns_in_a_row_share_a_role(db, speaker_id):
+    """The property, for every speaker in a three-model round. A run of assistant
+    turns is what a crowd produces naturally and what templates refuse."""
+    chat = _chat(db)
+    _round(
+        db,
+        chat,
+        [("main-model", "One."), ("second-model", "Two."), ("third-model", "Three.")],
+    )
+
+    roles = _roles(_payload(db, chat, speaker_id))
+
+    assert all(a != b for a, b in zip(roles, roles[1:], strict=False)), roles
+
+
+def test_the_history_still_starts_on_a_user_turn(db):
+    """What every chat template expects, and what the compaction pair exists to
+    preserve."""
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "One."), ("second-model", "Two.")])
+    assert _roles(_payload(db, chat, "third-model"))[0] == ROLE_USER
+
+
+def test_a_relabelled_turn_never_becomes_multimodal(db):
+    """Built directly rather than by calling `message_payload` with a swapped
+    role: that one attaches image parts when the role is `user`, so a swapped
+    assistant turn carrying a generated image would silently become a content
+    list -- and an endpoint that rejects one rejects every later turn with it."""
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "Here is a picture.")])
+
+    messages = _payload(db, chat, "second-model")
+
+    for entry in messages:
+        assert isinstance(entry["content"], str), entry
+
+
+# --- The instruction ----------------------------------------------------------
+def _turn(phase, index=1, of=3):
+    return crowd_service.Turn(
+        turn="u1", round=1, phase=phase, index=index, of=of,
+        started_at=crowd_service.now_stamp(),
+    )
+
+
+def test_the_forward_pass_asks_for_what_is_missing(db):
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "One.")])
+    messages = _payload(db, chat, "second-model", turn=_turn(crowd_service.PHASE_OUT))
+    assert "Add what is missing" in messages[-1]["content"]
+    assert messages[-1]["role"] == ROLE_USER
+
+
+def test_the_way_back_asks_for_disagreement(db):
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "One."), ("second-model", "Two.")])
+    messages = _payload(db, chat, "second-model", turn=_turn(crowd_service.PHASE_BACK))
+    assert "disagree" in messages[-1]["content"]
+
+
+def test_the_closing_turn_offers_another_round_only_when_there_is_one(db):
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "One."), ("second-model", "Two.")])
+
+    with_tool = _payload(
+        db, chat, "main-model", turn=_turn(crowd_service.PHASE_CLOSE, index=0), again=True
+    )
+    assert "crowd_again" in with_tool[-1]["content"]
+
+    without = _payload(
+        db, chat, "main-model", turn=_turn(crowd_service.PHASE_CLOSE, index=0), again=False
+    )
+    assert "crowd_again" not in without[-1]["content"]
+    assert "no further round" in without[-1]["content"]
+
+
+def test_the_instruction_is_not_written_into_the_transcript(db):
+    """Payload only. A row would double the bubbles, would be answered by every
+    later speaker as an ordinary user turn, and could be dropped from the request
+    entirely by a `created_at` tie with the placeholder."""
+    from lembas.db.models import Message
+
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "One.")])
+    before = db.scalar(select(Message).order_by(Message.created_at.desc()))
+
+    _payload(db, chat, "second-model", turn=_turn(crowd_service.PHASE_OUT))
+
+    db.expire_all()
+    rows = db.scalars(select(Message).where(Message.chat_id == chat.id)).all()
+    assert not any(
+        "Add what is missing" in (row.content or "") for row in rows
+    ), "the instruction was written into the conversation"
+    assert before is not None
+
+
+def test_clearing_the_fragment_sends_no_instruction(db):
+    """An administrator emptying a fragment is switching that wording off, which
+    is the convention everywhere else here -- and an empty user turn is not a
+    thing to send."""
+    prompts_service.save(db, {"crowd.turn": ""})
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "One.")])
+
+    messages = _payload(db, chat, "second-model", turn=_turn(crowd_service.PHASE_OUT))
+
+    assert messages[-1]["role"] == ROLE_USER
+    assert "Add what is missing" not in messages[-1]["content"]
+
+
+def test_the_instruction_merges_rather_than_doubling_a_user_turn(db):
+    """It lands after a quoted answer, which is itself a user turn now."""
+    chat = _chat(db)
+    _round(db, chat, [("main-model", "One.")])
+    roles = _roles(_payload(db, chat, "second-model", turn=_turn(crowd_service.PHASE_OUT)))
+    assert all(a != b for a, b in zip(roles, roles[1:], strict=False)), roles

tests/test_crowd_schedule.py

@@ -0,0 +1,231 @@
+"""The crowd's order of speaking, as arithmetic.
+
+`crowd.next_turn` is a pure function so that the interesting half of this feature
+— every way a round refuses to continue — can be tested without an endpoint, a
+session or a clock. The order the owner asked for is one sequence, and getting it
+wrong in either direction is a feature that looks like it works: a backward pass
+that starts on the speaker who has just spoken asks it whether it disagrees with
+itself, and one that runs to the main model twice gives it two closing turns.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from lembas.services import crowd
+
+
+def _first(speakers: int) -> crowd.Turn:
+    turn = crowd.next_turn(speakers=speakers, state=None, turn_id="u1")
+    assert turn is not None
+    return turn
+
+
+def _walk(speakers: int, *, again_at: set[int] = frozenset(), max_rounds: int = 2) -> list[str]:
+    """The whole sequence as `phase/index` strings, for one readable assertion."""
+    state = None
+    seen: list[str] = []
+    for _ in range(60):
+        again = state is not None and state.round in again_at and state.phase == crowd.PHASE_CLOSE
+        turn = crowd.next_turn(
+            speakers=speakers,
+            state=state,
+            turn_id="u1",
+            again=again,
+            max_rounds=max_rounds,
+        )
+        if turn is None or turn.stopped:
+            if turn is not None and turn.stopped:
+                seen.append(f"stopped:{turn.stopped}")
+            break
+        seen.append(f"{turn.phase}/{turn.index}")
+        state = turn
+    return seen
+
+
+# --- The order ----------------------------------------------------------------
+def test_one_model_is_not_a_crowd():
+    """The chat's own model with nobody else answers exactly as it always did."""
+    assert crowd.next_turn(speakers=1, state=None, turn_id="u1") is None
+
+
+def test_two_speakers_go_out_and_straight_back_to_the_main_model():
+    """With one member there is nobody to ask on the way back, so the round is
+    main, member, main — and the backward pass is empty rather than asking the
+    member about its own answer."""
+    assert _walk(2) == ["out/1", "close/0"]
+
+
+def test_three_speakers_come_back_through_the_middle():
+    assert _walk(3) == ["out/1", "out/2", "back/1", "close/0"]
+
+
+def test_five_speakers_walk_out_and_back_in_order():
+    assert _walk(5) == [
+        "out/1", "out/2", "out/3", "out/4",
+        "back/3", "back/2", "back/1",
+        "close/0",
+    ]
+
+
+def test_the_way_back_never_asks_the_last_speaker_about_itself():
+    """It starts one short of the speaker that has just finished."""
+    for speakers in range(2, 7):
+        sequence = _walk(speakers)
+        out = [s for s in sequence if s.startswith("out/")]
+        back = [s for s in sequence if s.startswith("back/")]
+        if back:
+            assert back[0] != out[-1].replace("out/", "back/")
+
+
+def test_the_main_model_gets_exactly_one_closing_turn():
+    for speakers in range(2, 7):
+        assert _walk(speakers).count("close/0") == 1
+
+
+def test_the_first_reply_is_not_scheduled_by_this():
+    """The composer starts it, as it always has. A round *begins* at the second
+    speaker, which is why `state=None` returns index 1."""
+    assert _first(4).index == 1
+    assert _first(4).phase == crowd.PHASE_OUT
+    assert _first(4).round == 1
+
+
+def test_the_size_of_the_round_is_recorded_on_every_turn():
+    """`of` is what the chip in the transcript counts against."""
+    turn = _first(4)
+    assert turn.of == 4
+
+
+# --- Going round again ---------------------------------------------------------
+def test_without_being_asked_the_round_ends_at_the_main_model():
+    assert _walk(3, again_at=set()) == ["out/1", "out/2", "back/1", "close/0"]
+
+
+def test_asked_for_another_round_it_starts_again_at_the_second_speaker():
+    """The main model has just spoken as the closer, so round two begins with the
+    others rather than with it."""
+    sequence = _walk(3, again_at={1}, max_rounds=2)
+    assert sequence == [
+        "out/1", "out/2", "back/1", "close/0",
+        "out/1", "out/2", "back/1", "close/0",
+    ]
+
+
+def test_the_round_cap_stops_it_and_says_why():
+    """Reached rather than never: the cap is a ceiling on ordinary work here,
+    unlike a runaway backstop, so somebody has to be able to see it was hit."""
+    sequence = _walk(3, again_at={1, 2, 3}, max_rounds=2)
+    assert sequence[-1] == f"stopped:{crowd.STOPPED_ROUNDS}"
+    assert sequence.count("close/0") == 2
+
+
+def test_one_round_means_one_round():
+    sequence = _walk(3, again_at={1, 2}, max_rounds=1)
+    assert sequence.count("close/0") == 1
+    assert sequence[-1] == f"stopped:{crowd.STOPPED_ROUNDS}"
+
+
+# --- Running out of time -------------------------------------------------------
+def _stale(seconds: int) -> crowd.Turn:
+    began = datetime.now(UTC) - timedelta(seconds=seconds)
+    return crowd.Turn(
+        turn="u1", round=1, phase=crowd.PHASE_OUT, index=1, of=4,
+        started_at=began.isoformat(),
+    )
+
+
+def test_a_round_that_has_run_long_enough_is_stopped():
+    stopped = crowd.next_turn(speakers=4, state=_stale(1000), turn_id="u1", wall_seconds=900)
+    assert stopped is not None
+    assert stopped.stopped == crowd.STOPPED_TIME
+
+
+def test_a_round_inside_its_time_carries_on():
+    turn = crowd.next_turn(speakers=4, state=_stale(10), turn_id="u1", wall_seconds=900)
+    assert turn is not None
+    assert not turn.stopped
+    assert turn.index == 2
+
+
+def test_the_clock_covers_the_whole_turn_not_one_speaker():
+    """`started_at` is carried from the round's first turn, never refreshed, so a
+    crowd of slow members cannot outrun the limit one speaker at a time."""
+    first = _first(4)
+    second = crowd.next_turn(speakers=4, state=first, turn_id="u1")
+    assert second is not None
+    assert second.started_at == first.started_at
+
+
+def test_an_unreadable_stamp_reads_as_no_time_passed():
+    """A round abandoned because of a bad timestamp would be a feature failing
+    for a reason nobody could see."""
+    broken = crowd.Turn(
+        turn="u1", round=1, phase=crowd.PHASE_OUT, index=1, of=4, started_at="not a date"
+    )
+    turn = crowd.next_turn(speakers=4, state=broken, turn_id="u1", wall_seconds=1)
+    assert turn is not None
+    assert not turn.stopped
+
+
+# --- Errors -------------------------------------------------------------------
+def test_one_speaker_failing_is_skipped_rather_than_ending_the_round():
+    """The commonest failure is a small member's window overflowing on a
+    transcript several models have written into. Ending the round there would kill
+    every crowd at whichever member is smallest."""
+    turn = crowd.next_turn(speakers=5, state=_first(5), turn_id="u1", errored=True)
+    assert turn is not None
+    assert not turn.stopped
+    assert turn.index == 2
+    assert turn.errors == 1
+
+
+def test_two_failures_in_a_row_end_the_round():
+    """Which is `_drain`'s protection kept: the endpoint has actually gone, and
+    feeding it the next prompt produces a second failure and spends the words to
+    do it."""
+    first = crowd.next_turn(speakers=5, state=_first(5), turn_id="u1", errored=True)
+    second = crowd.next_turn(speakers=5, state=first, turn_id="u1", errored=True)
+    assert second is not None
+    assert second.stopped == crowd.STOPPED_ERRORS
+
+
+def test_the_count_is_of_consecutive_failures():
+    """One failure, then a success, then a failure is not a dead endpoint."""
+    state = crowd.next_turn(speakers=6, state=_first(6), turn_id="u1", errored=True)
+    assert state.errors == 1
+    state = crowd.next_turn(speakers=6, state=state, turn_id="u1", errored=False)
+    assert state.errors == 0
+    state = crowd.next_turn(speakers=6, state=state, turn_id="u1", errored=True)
+    assert state is not None
+    assert not state.stopped
+
+
+# --- What is stored -----------------------------------------------------------
+def test_the_state_survives_a_round_trip_through_the_row():
+    """It is read back off a message after a restart, so the two halves have to
+    agree exactly."""
+
+    class Row:
+        crowd_json = None
+
+    turn = _first(4)
+    Row.crowd_json = turn.as_json()
+    assert crowd.state_of(Row) == turn
+
+
+def test_a_message_with_no_state_is_not_part_of_a_round():
+    class Row:
+        crowd_json = None
+
+    assert crowd.state_of(Row) is None
+    assert crowd.state_of(None) is None
+
+
+def test_nonsense_on_the_row_reads_as_no_round():
+    """A hand-edited database must not raise inside the generation loop."""
+
+    class Row:
+        crowd_json = {"round": "third", "index": None}
+
+    assert crowd.state_of(Row) is None

tests/test_crowd_ui.py

@@ -0,0 +1,242 @@
+"""Choosing a crowd, and reading one.
+
+Two screens and one rule each. The picker may only ever offer and accept models
+*this person* can reach — a control checked in the template and not in the route is
+advisory, and a crafted request walks past it. The transcript has to say which
+speaker a bubble is and which pass it belongs to, because nine bubbles for one
+question are otherwise indistinguishable from nine people talking at once.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import select
+
+from lembas.db.models import (
+    ROLE_ASSISTANT,
+    ROLE_USER,
+    Chat,
+    Connection,
+    Group,
+    Model,
+    User,
+)
+from lembas.services import chat as chat_service
+from lembas.services import crowd as crowd_service
+from lembas.services import settings_store
+from lembas.services.crypto import encrypt
+
+
+@pytest.fixture(autouse=True)
+def crowd_on(db, registered):
+    settings_store.update(db, {"enabled": True}, key=settings_store.CROWD)
+    connection = Connection(
+        name="Test", base_url="http://127.0.0.1:1", api_key_encrypted=encrypt("")
+    )
+    db.add(connection)
+    db.commit()
+    for index, name in enumerate(("main-model", "second-model", "third-model")):
+        db.add(
+            Model(
+                connection_id=connection.id,
+                model_id=name,
+                display_name=name,
+                position=index,
+                capabilities_json={"tools": True},
+            )
+        )
+    db.commit()
+
+
+def _user(db) -> User:
+    return db.scalars(select(User).order_by(User.created_at)).first()
+
+
+def _chat(db) -> Chat:
+    connection = db.scalars(select(Connection)).first()
+    chat = Chat(
+        user_id=_user(db).id, title="t", model_id="main-model", connection_id=connection.id
+    )
+    db.add(chat)
+    db.commit()
+    return chat
+
+
+def _members(db, chat) -> list[str]:
+    db.expire_all()
+    return [
+        row.model_id
+        for row in sorted(db.get(Chat, chat.id).crowd, key=lambda r: r.position)
+    ]
+
+
+# --- Choosing -----------------------------------------------------------------
+def test_the_panel_offers_the_other_models(client, db):
+    chat = _chat(db)
+    page = client.get(f"/chat/{chat.id}").text
+    assert 'name="crowd_model_ids"' in page
+    assert 'name="crowd_model_ids" value="second-model"' in page
+    # Never the chat's own model: it would answer twice in a row. Asserted with
+    # the field name attached, because the model *picker* on the same page quite
+    # correctly offers it.
+    assert 'name="crowd_model_ids" value="main-model"' not in page
+
+
+def test_the_panel_is_absent_while_the_feature_is_off(client, db):
+    settings_store.update(db, {"enabled": False}, key=settings_store.CROWD)
+    chat = _chat(db)
+    assert 'name="crowd_model_ids"' not in client.get(f"/chat/{chat.id}").text
+
+
+def test_ticking_a_model_adds_it_in_order(client, db):
+    chat = _chat(db)
+    client.patch(
+        f"/api/chats/{chat.id}",
+        data={"crowd_model_ids": ["second-model", "third-model"]},
+    )
+    assert _members(db, chat) == ["second-model", "third-model"]
+
+
+def test_clearing_every_box_clears_the_crowd(client, db):
+    """The single field always sent is what makes this possible: an absent
+    checkbox carries no signal of its own."""
+    chat = _chat(db)
+    client.patch(f"/api/chats/{chat.id}", data={"crowd_model_ids": ["second-model"]})
+    assert _members(db, chat) == ["second-model"]
+
+    client.patch(f"/api/chats/{chat.id}", data={"crowd_model_ids": [""]})
+    assert _members(db, chat) == []
+
+
+def test_a_model_this_person_cannot_reach_is_refused(client, db):
+    """Checked in the route, not only in the template. Otherwise the picker is
+    advisory."""
+    group = Group(name="Wheel")
+    db.add(group)
+    restricted = db.scalar(select(Model).where(Model.model_id == "third-model"))
+    restricted.public = False
+    restricted.groups = [group]
+    user = _user(db)
+    user.role = "user"
+    db.commit()
+    chat = _chat(db)
+
+    client.patch(
+        f"/api/chats/{chat.id}",
+        data={"crowd_model_ids": ["second-model", "third-model"]},
+    )
+
+    assert _members(db, chat) == ["second-model"]
+
+
+def test_the_chats_own_model_cannot_be_added(client, db):
+    chat = _chat(db)
+    client.patch(f"/api/chats/{chat.id}", data={"crowd_model_ids": ["main-model"]})
+    assert _members(db, chat) == []
+
+
+def test_the_same_model_twice_is_one_member(client, db):
+    chat = _chat(db)
+    client.patch(
+        f"/api/chats/{chat.id}", data={"crowd_model_ids": ["second-model", "second-model"]}
+    )
+    assert _members(db, chat) == ["second-model"]
+
+
+def test_the_cap_trims_what_is_accepted(client, db):
+    settings_store.update(db, {"max_models": 1}, key=settings_store.CROWD)
+    chat = _chat(db)
+    client.patch(
+        f"/api/chats/{chat.id}", data={"crowd_model_ids": ["second-model", "third-model"]}
+    )
+    assert _members(db, chat) == ["second-model"]
+
+
+def test_the_panel_says_what_a_turn_will_cost(client, db):
+    """The thing somebody will not have thought about: a turn is
+    speakers x rounds x 2 - 1 replies, and each is a whole reply."""
+    chat = _chat(db)
+    client.patch(
+        f"/api/chats/{chat.id}", data={"crowd_model_ids": ["second-model", "third-model"]}
+    )
+    page = client.get(f"/chat/{chat.id}").text
+    assert "5 replies a turn" in page
+
+
+def test_a_member_that_can_no_longer_be_reached_is_shown_struck_through(client, db):
+    """Membership is text with no foreign key, so the row outlives the model. Saying
+    so beats both deleting it and pretending it still answers."""
+    chat = _chat(db)
+    client.patch(f"/api/chats/{chat.id}", data={"crowd_model_ids": ["third-model"]})
+    gone = db.scalar(select(Model).where(Model.model_id == "third-model"))
+    db.delete(gone)
+    db.commit()
+
+    page = client.get(f"/chat/{chat.id}").text
+
+    assert "Skipped" in page
+    assert "<s>third-model</s>" in page
+
+
+# --- Reading ------------------------------------------------------------------
+def _bubble(db, chat, *, phase, index=1, of=3, stopped="", round_=1) -> str:
+    from lembas.api import chats as chats_api
+
+    chat_service.create_message(db, chat, ROLE_USER, "What should we do?")
+    message = chat_service.create_message(
+        db, chat, ROLE_ASSISTANT, "Something.", model_id="second-model"
+    )
+    message.crowd_json = crowd_service.Turn(
+        turn="u1", round=round_, phase=phase, index=index, of=of,
+        started_at=crowd_service.now_stamp(), stopped=stopped,
+    ).as_json()
+    db.commit()
+    return chats_api._render_bubble(db, chat, _user(db), message)
+
+
+def test_a_bubble_on_the_way_out_says_which_speaker_it_is(db):
+    chat = _chat(db)
+    html = _bubble(db, chat, phase=crowd_service.PHASE_OUT, index=1, of=3)
+    assert "2 of 3" in html
+
+
+def test_a_bubble_on_the_way_back_says_so_and_is_quieter(db):
+    chat = _chat(db)
+    html = _bubble(db, chat, phase=crowd_service.PHASE_BACK)
+    assert "on the way back" in html
+    assert "msg--crowd-back" in html
+
+
+def test_the_closing_bubble_says_it_is_closing(db):
+    chat = _chat(db)
+    html = _bubble(db, chat, phase=crowd_service.PHASE_CLOSE, index=0)
+    assert "closing" in html
+
+
+def test_a_later_round_is_numbered(db):
+    chat = _chat(db)
+    html = _bubble(db, chat, phase=crowd_service.PHASE_OUT, round_=2)
+    assert "round 2" in html
+
+
+def test_why_a_round_ended_is_shown_where_it_ended(db):
+    """Otherwise a crowd that ran out of rounds or time simply stops, which reads
+    as the feature failing rather than as a limit doing its job."""
+    chat = _chat(db)
+    assert "no rounds left" in _bubble(
+        db, chat, phase=crowd_service.PHASE_CLOSE, stopped=crowd_service.STOPPED_ROUNDS
+    )
+
+
+def test_an_ordinary_bubble_carries_no_crowd_chip(db):
+    from lembas.api import chats as chats_api
+
+    chat = _chat(db)
+    chat_service.create_message(db, chat, ROLE_USER, "Hello")
+    message = chat_service.create_message(
+        db, chat, ROLE_ASSISTANT, "Hello back.", model_id="main-model"
+    )
+    html = chats_api._render_bubble(db, chat, _user(db), message)
+    assert "of 3" not in html
+    assert "msg--crowd-back" not in html
+    assert "closing" not in html

tests/test_migrations.py

@@ -36,6 +36,7 @@ OLD_TABLES = (
     "personas",
     "persona_revisions",
     "impressions",
+    "chat_crowd",
 )
 
 # Columns added to tables that already existed, and therefore already had rows.
@@ -53,6 +54,11 @@ OLD_COLUMNS = (
     # live instance, and a column absent from this list is a column the migration
     # tests do not exercise.
     ("models", "notes"),
+    # Which model wrote a message, and where it sits in a crowd round. Both
+    # nullable, so the backfill is the easy kind -- listed because a column absent
+    # from here is one the migration tests do not exercise at all.
+    ("messages", "connection_id"),
+    ("messages", "crowd_json"),
 )
 
 

tests/test_speaker.py

@@ -0,0 +1,292 @@
+"""Which model answers one reply, and where that is decided.
+
+Until 1.6.0 it was `chat.model_id` and nothing else, while `Message.model_id` was
+written on every assistant placeholder and read only for display. The two could
+disagree, and did: `wake_chat` accepts a `model_id` override, `schedule/runner`
+passes `schedule.model_id or chat.model_id`, and that reached the row and never
+reached the request — so a schedule naming another model got the chat's model
+wearing the other one's name on the bubble. Half a feature, wired and unread.
+
+The row is the authority now. That is also what makes a reply survive a restart:
+`_follow` calls `ensure`, which starts a **new** generation against the same row,
+so anything the request depends on has to be durable — and the in-process registry
+is not.
+
+Everything that differs per model is asserted here, because each of them fails
+differently and three of them fail silently:
+
+* the model id sent, which is the visible one;
+* `vision`, where a wrong answer makes the endpoint reject the **whole request**;
+* the reasoning-effort vocabulary, which raises inside the model's chat template;
+* the tools capability, `context_length`, `{{model_name}}`, the personality, and
+  the authored prompt's model layer.
+"""
+
+from __future__ import annotations
+
+import pytest
+from sqlalchemy import select
+
+from lembas.db.models import ROLE_ASSISTANT, ROLE_USER, Chat, Connection, Model, User
+from lembas.services import chat as chat_service
+from lembas.services import harness as harness_service
+from lembas.services import personas as personas_service
+from lembas.services import settings_store
+from lembas.services import tools as tools_service
+from lembas.services.crypto import encrypt
+
+
+@pytest.fixture(autouse=True)
+def two_models(db, registered):
+    connection = Connection(
+        name="Test", base_url="http://127.0.0.1:1", api_key_encrypted=encrypt("")
+    )
+    db.add(connection)
+    db.commit()
+    db.add(
+        Model(
+            connection_id=connection.id,
+            model_id="the-chats-model",
+            display_name="Chat model",
+            position=0,
+            context_length=8192,
+            reasoning_efforts=["low", "medium", "high"],
+            system_prompt="You are the chat's model.",
+            capabilities_json={"tools": True, "vision": True},
+        )
+    )
+    db.add(
+        Model(
+            connection_id=connection.id,
+            model_id="the-other-model",
+            display_name="Other model",
+            position=1,
+            context_length=128000,
+            reasoning_efforts=["low", "medium", "xhigh"],
+            system_prompt="You are the other model.",
+            capabilities_json={"tools": False, "vision": False},
+        )
+    )
+    db.commit()
+
+
+def _user(db) -> User:
+    return db.scalars(select(User).order_by(User.created_at)).first()
+
+
+def _chat(db) -> Chat:
+    connection = db.scalars(select(Connection)).first()
+    chat = Chat(
+        user_id=_user(db).id,
+        title="t",
+        model_id="the-chats-model",
+        connection_id=connection.id,
+    )
+    db.add(chat)
+    db.commit()
+    return chat
+
+
+def _turn(db, chat, *, model_id: str = ""):
+    """A user turn and the assistant placeholder that answers it."""
+    chat_service.create_message(db, chat, ROLE_USER, "Say something")
+    return chat_service.create_message(
+        db, chat, ROLE_ASSISTANT, "", complete_=False, model_id=model_id or chat.model_id
+    )
+
+
+# --- Where it is decided ------------------------------------------------------
+def test_the_row_names_the_model_and_the_chat_is_the_default(db):
+    chat = _chat(db)
+    assert chat_service.speaker_for(db, chat).model_id == "the-chats-model"
+
+    placeholder = _turn(db, chat, model_id="the-other-model")
+    assert chat_service.speaker_for(db, chat, placeholder).model_id == "the-other-model"
+
+
+def test_a_row_naming_no_model_falls_back_to_the_chat(db):
+    """Every existing row names one, but a row written by an older release or by
+    some future caller that forgot must not send an empty model id."""
+    chat = _chat(db)
+    placeholder = _turn(db, chat)
+    placeholder.model_id = ""
+    db.commit()
+    assert chat_service.speaker_for(db, chat, placeholder).model_id == "the-chats-model"
+
+
+# --- What the request carries -------------------------------------------------
+def test_the_request_is_sent_to_the_model_the_row_names(db):
+    """The bug, in one assertion. This failed before the speaker existed."""
+    chat = _chat(db)
+    placeholder = _turn(db, chat, model_id="the-other-model")
+
+    body = chat_service.build_request(db, chat, upto=placeholder, user=_user(db))
+
+    assert body["model"] == "the-other-model"
+
+
+def test_the_endpoint_is_resolved_for_the_row_s_model(db):
+    chat = _chat(db)
+    placeholder = _turn(db, chat, model_id="the-other-model")
+    speaker = chat_service.speaker_for(db, chat, placeholder)
+
+    _endpoint, model_id = chat_service.resolve_endpoint(db, chat, speaker)
+
+    assert model_id == "the-other-model"
+
+
+def test_resolving_another_model_s_connection_does_not_repoint_the_chat(db):
+    """`resolve_endpoint` writes `chat.connection_id` when the original has gone.
+    For a speaker that is not the chat's own model that would quietly move the
+    whole conversation to another endpoint."""
+    chat = _chat(db)
+    original = chat.connection_id
+    second = Connection(name="Second", base_url="http://127.0.0.2:1", api_key_encrypted=encrypt(""))
+    db.add(second)
+    db.commit()
+    db.add(Model(connection_id=second.id, model_id="only-here", position=9))
+    db.commit()
+
+    chat_service.resolve_endpoint(db, chat, chat_service.Speaker("only-here", None))
+
+    db.expire_all()
+    assert db.get(Chat, chat.id).connection_id == original
+
+
+def test_the_effort_vocabulary_is_the_answering_model_s(db):
+    """Not cosmetic: an effort a model does not take is rendered into its chat
+    template and raises there, failing the whole reply. gpt-oss takes
+    low/medium/high; a Bonsai takes low/medium/xhigh and refuses high."""
+    chat = _chat(db)
+    chat.params_json = {"reasoning_effort": "high"}
+    db.commit()
+    placeholder = _turn(db, chat, model_id="the-other-model")
+
+    body = chat_service.build_request(db, chat, upto=placeholder, user=_user(db))
+
+    # `high` is not in the other model's list, so it is not sent at all rather
+    # than being sent to a template that raises on it.
+    assert body.get("reasoning_effort") != "high"
+    kwargs = body.get("chat_template_kwargs") or {}
+    assert kwargs.get("reasoning_effort") != "high"
+
+
+def test_an_effort_the_answering_model_does_take_is_sent(db):
+    chat = _chat(db)
+    chat.params_json = {"reasoning_effort": "medium"}
+    db.commit()
+    placeholder = _turn(db, chat, model_id="the-other-model")
+
+    body = chat_service.build_request(db, chat, upto=placeholder, user=_user(db))
+
+    assert body["reasoning_effort"] == "medium"
+
+
+def test_vision_follows_the_answering_model(db):
+    """An image sent to a model without vision is not degraded gracefully: most
+    endpoints reject the entire request."""
+    chat = _chat(db)
+    assert chat_service.model_supports(db, chat, "vision") is True
+    assert (
+        chat_service.model_supports(
+            db, chat, "vision", speaker=chat_service.Speaker("the-other-model")
+        )
+        is False
+    )
+
+
+def test_the_authored_prompt_uses_the_answering_model_s_layer(db):
+    chat = _chat(db)
+    speaker = chat_service.Speaker("the-other-model")
+
+    assert "chat's model" in chat_service.effective_system_prompt(db, chat)
+    assert "other model" in chat_service.effective_system_prompt(db, chat, speaker)
+
+
+def test_the_tools_capability_is_the_answering_model_s(db):
+    """`tools` off is the first gate and returns nothing at all, so a model that
+    cannot take a tools array must not be handed one -- its replies fail rather
+    than degrade."""
+    chat = _chat(db)
+    user = _user(db)
+    settings_store.update(db, {"default_permissions": {"tools.web_search": True}})
+
+    assert tools_service.resolve_tools(db, chat, user).defs
+    assert not tools_service.resolve_tools(
+        db, chat, user, chat_service.Speaker("the-other-model")
+    ).defs
+
+
+def test_the_context_limit_is_the_answering_model_s(db):
+    chat = _chat(db)
+    assert chat_service.model_for(db, chat).context_length == 8192
+    other = chat_service.model_row(db, chat_service.Speaker("the-other-model"))
+    assert other.context_length == 128000
+
+
+def test_the_model_name_variable_is_the_answering_model_s(db):
+    """Telling a speaker it is the main model is a lie it then reasons from."""
+    chat = _chat(db)
+    values = harness_service.context_variables(
+        db, _user(db), [], chat, chat_service.Speaker("the-other-model")
+    )
+    assert values["model_name"] == "Other model"
+
+
+def test_the_personality_is_the_answering_model_s(db):
+    chat = _chat(db)
+    user = _user(db)
+    settings_store.update(db, {"default_permissions": {"tools.persona": True}})
+    personas_service.write(db, model_key="the-chats-model", owner=user, content="I am the chat's.")
+    personas_service.write(db, model_key="the-other-model", owner=user, content="I am the other.")
+
+    offered = [
+        tool.schema
+        for tool in tools_service.registry(db).values()
+        if tools_service.gate_of(tool.family) == "persona"
+    ]
+    mine = harness_service.context_variables(db, user, offered, chat)
+    theirs = harness_service.context_variables(
+        db, user, offered, chat, chat_service.Speaker("the-other-model")
+    )
+
+    assert mine["persona"] == "I am the chat's."
+    assert theirs["persona"] == "I am the other."
+
+
+def test_a_tool_acts_as_the_answering_model(db):
+    """`ToolContext.model_id` is which model a tool acts *as* -- whose personality
+    `persona_write` rewrites, and whose endpoint the image reviewer reaches for."""
+    chat = _chat(db)
+    context = tools_service.context_for(
+        db, _user(db), chat, speaker=chat_service.Speaker("the-other-model", "abc")
+    )
+    assert context.model_id == "the-other-model"
+    assert context.connection_id == "abc"
+
+
+# --- The half-wired feature this closes ---------------------------------------
+async def test_a_schedule_naming_another_model_now_sends_it(db, monkeypatch):
+    """`wake_chat(model_id=…)` wrote the override onto the row and `_run` ignored
+    it. End to end: the turn goes in through the documented path, and the request
+    built for the placeholder it created names the model the caller asked for."""
+    from lembas.services import generation as generation_service
+    from lembas.services import wake as wake_service
+
+    chat = _chat(db)
+    monkeypatch.setattr(generation_service, "running_for", lambda chat_id: None)
+    monkeypatch.setattr(generation_service, "ensure", lambda chat_id, message_id: None)
+
+    message_id = await wake_service.wake_chat(
+        chat.id, "Run the nightly summary", model_id="the-other-model"
+    )
+
+    db.expire_all()
+    from lembas.db.models import Message
+
+    placeholder = db.get(Message, message_id)
+    assert placeholder.model_id == "the-other-model"
+    body = chat_service.build_request(
+        db, db.get(Chat, chat.id), upto=placeholder, user=_user(db)
+    )
+    assert body["model"] == "the-other-model"