diff --git a/tests/test_agent_interaction.py b/tests/test_agent_interaction.py
index d741027..90959e5 100644
--- a/tests/test_agent_interaction.py
+++ b/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(
diff --git a/tests/test_chat.py b/tests/test_chat.py
index 70a0a78..68f3639 100644
--- a/tests/test_chat.py
+++ b/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.
- assert remaining[1].complete is False
+ # A *fresh* assistant row, which is what restarts the stream: a different row
+ # 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(
diff --git a/tests/test_crowd_chain.py b/tests/test_crowd_chain.py
new file mode 100644
index 0000000..6932fe0
--- /dev/null
+++ b/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
diff --git a/tests/test_crowd_payload.py b/tests/test_crowd_payload.py
new file mode 100644
index 0000000..6105cf3
--- /dev/null
+++ b/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
diff --git a/tests/test_crowd_schedule.py b/tests/test_crowd_schedule.py
new file mode 100644
index 0000000..a668b0f
--- /dev/null
+++ b/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
diff --git a/tests/test_crowd_ui.py b/tests/test_crowd_ui.py
new file mode 100644
index 0000000..557acbb
--- /dev/null
+++ b/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 "third-model" 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
diff --git a/tests/test_migrations.py b/tests/test_migrations.py
index 001ed9b..5aec211 100644
--- a/tests/test_migrations.py
+++ b/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"),
)
diff --git a/tests/test_speaker.py b/tests/test_speaker.py
new file mode 100644
index 0000000..fcf80a6
--- /dev/null
+++ b/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"