fix(session): properly inject queued user messages mid-loop

Two queued-user-message bugs in ``ChatSession.send()``.

**Mid-tool-call: ``Unexpected role 'tool' after role 'user'`` on Mistral.**
The ``supports_tool_advisories`` capability flag (default False for
unknown openai-compatible models) routed cap-off providers down a
short-circuit branch in ``_collect_advisories`` that called
``_flush_queued_messages`` directly. That appended a ``user`` turn
between ``assistant(tool_calls)`` and ``tool``, which mistral-common's
``_validate_message_order`` rejects with a 400.

Drop the flag. All providers now run the unified path: queued user
messages become ``UserInterjection`` advisories that ride inside the
tool result envelope via ``wrap_tool_result``, splicing
``<system-reminder>`` text into the tool message's content. Role
sequence stays ``assistant → tool``. Live-confirmed on Mistral
medium and Qwen3 — both correctly distinguish system-reminder from
tool stdout in their reasoning.

**Mid-stream: queued message orphaned until next user send.**
After a no-tool assistant turn, ``_flush_queued_messages`` would
append the queued user message to history and the loop would
``break``, leaving the message at the tail of history with no
model response. Visible as "two sends to get one reply".

``_flush_queued_messages`` now returns ``bool``. The no-tool branch
``continue``s on drain instead of ``break``ing, so the model gets a
turn over the extended history.

Tests:
- ``test_collect_advisories_drains_text_queued_messages_to_persistent``
  pins the unified-path drain (text-only queue → ``UserInterjection``,
  no separate user turn appended to ``self.messages``).
- ``test_send_continues_when_messages_queued_during_streaming`` pins
  the loop-continue behavior (fails with 1 stream call pre-fix,
  passes with 2 post-fix).

tests/test_session.py

@@ -1984,13 +1984,6 @@ class TestMetacognitiveBuffers:
         assert session._pending_user_advisories == [("correction", "USER_NUDGE_MARK")]
         assert session._pending_tool_advisories == [("tool_error", "TOOL_NUDGE_MARK")]
 
-    def _patch_caps(self, session, *, supports_tool_advisories: bool):
-        """Force capability flag for advisory-aware tests."""
-        caps = MagicMock()
-        caps.supports_tool_advisories = supports_tool_advisories
-        with patch.object(session, "_get_capabilities", return_value=caps):
-            return caps
-
     def test_collect_advisories_drains_tool_buffer_on_last_result(self, tmp_db):
         """Tool-channel metacog reminders no longer ride the persistent
         advisory list (which would write them into tool content via
@@ -2000,12 +1993,9 @@ class TestMetacognitiveBuffers:
         channel."""
         session = _make_session()
         session._queue_tool_advisory("tool_error", "ALERT")
-        caps = MagicMock()
-        caps.supports_tool_advisories = True
-        with patch.object(session, "_get_capabilities", return_value=caps):
-            persistent, metacog = session._collect_advisories(
-                assessment=None, func_name="bash", is_last_in_batch=True
-            )
+        persistent, metacog = session._collect_advisories(
+            assessment=None, func_name="bash", is_last_in_batch=True
+        )
         # Persistent list is empty (no guard / interjection here);
         # MetacognitiveAdvisory does NOT appear among persistent
         # advisories anymore.
@@ -2017,32 +2007,41 @@ class TestMetacognitiveBuffers:
     def test_collect_advisories_holds_tool_buffer_until_last_result(self, tmp_db):
         session = _make_session()
         session._queue_tool_advisory("repeat", "STOP_REPEATING")
-        caps = MagicMock()
-        caps.supports_tool_advisories = True
-        with patch.object(session, "_get_capabilities", return_value=caps):
-            persistent, metacog = session._collect_advisories(
-                assessment=None, func_name="bash", is_last_in_batch=False
-            )
+        persistent, metacog = session._collect_advisories(
+            assessment=None, func_name="bash", is_last_in_batch=False
+        )
         # Not yet drained — only fires on the last result.
         assert persistent == []
         assert metacog == []
         assert len(session._pending_tool_advisories) == 1
 
-    def test_collect_advisories_drops_tool_buffer_when_caps_unsupported(self, tmp_db):
-        """When the model can't parse advisory tags, drop the metacognitive
-        nudge silently rather than embedding raw XML the model will choke on."""
+    def test_collect_advisories_drains_text_queued_messages_to_persistent(self, tmp_db):
+        """Text-only queued user messages drain into the ``persistent``
+        advisory list as ``UserInterjection`` on the last result of a
+        batch — they ride INSIDE the tool result envelope via
+        ``wrap_tool_result`` rather than becoming a separate user turn
+        appended to ``self.messages`` (which would inject ``user``
+        between ``assistant(tool_calls)`` and ``tool`` and break role
+        validation on Mistral / mistral-common and similar strict
+        templates)."""
+        from turnstone.core.tool_advisory import UserInterjection
+
         session = _make_session()
-        session._queue_tool_advisory("tool_error", "ALERT")
-        caps = MagicMock()
-        caps.supports_tool_advisories = False
-        with patch.object(session, "_get_capabilities", return_value=caps):
-            persistent, metacog = session._collect_advisories(
-                assessment=None, func_name="bash", is_last_in_batch=True
-            )
-        assert persistent == []
+        pre_count = len(session.messages)
+        session.queue_message("hows it going?", queue_msg_id="q1")
+        persistent, metacog = session._collect_advisories(
+            assessment=None, func_name="bash", is_last_in_batch=True
+        )
         assert metacog == []
-        # And the buffer is cleared so no stale nudge sticks around.
-        assert session._pending_tool_advisories == []
+        assert len(persistent) == 1
+        assert isinstance(persistent[0], UserInterjection)
+        assert persistent[0].message == "hows it going?"
+        # Queue drained.
+        assert session._queued_messages == {}
+        # Crucially: NO separate user turn was appended to history —
+        # the message rides inside the tool envelope, preserving the
+        # `assistant(tool_calls) → tool` role sequence on the wire.
+        assert len(session.messages) == pre_count
 
     def test_start_nudge_fires_through_send(self, tmp_db):
         """Pin the +1 count-shift invariant — `start` must still fire on the
@@ -2112,10 +2111,7 @@ class TestMetacognitiveBuffers:
         session = _make_session()
         session.ui = MagicMock()
         session._queue_tool_advisory("tool_error", "alert")
-        caps = MagicMock()
-        caps.supports_tool_advisories = True
-        with patch.object(session, "_get_capabilities", return_value=caps):
-            session._collect_advisories(assessment=None, func_name="bash", is_last_in_batch=True)
+        session._collect_advisories(assessment=None, func_name="bash", is_last_in_batch=True)
         info_lines = [call.args[0] for call in session.ui.on_info.call_args_list if call.args]
         assert not any("metacognition: nudge injected" in line for line in info_lines), (
             f"expected NO legacy ping, got {info_lines!r}"
@@ -2786,6 +2782,74 @@ class TestUserAdvisoryCancelClear:
             session.send("user input")
         assert session._pending_user_advisories == []
 
+    def test_send_continues_when_messages_queued_during_streaming(self, tmp_db):
+        """A user message queued while the assistant is streaming a
+        non-tool response must trigger another model turn — not orphan
+        in history until the next user send.
+
+        Pre-fix bug: after the no-tool branch ran ``_flush_queued_messages``,
+        the loop ``break``-d unconditionally, leaving the queued user
+        message at the tail of ``self.messages`` with no model response.
+        The next outside ``send()`` would finally pick it up alongside
+        the new message — visible as the "two sends to get one reply"
+        symptom.
+
+        Fix: ``_flush_queued_messages`` returns whether anything drained;
+        the no-tool branch ``continue``-s when it did."""
+        session = _make_session()
+        # Suppress the auto-title daemon thread the no-tool branch
+        # would spawn — irrelevant to this test and would otherwise
+        # call the mocked client from a background thread.
+        session._title_generated = True
+        stream_calls = 0
+
+        def mock_create_stream(msgs):
+            nonlocal stream_calls
+            stream_calls += 1
+            if stream_calls == 1:
+                # Simulate a queued message arriving mid-stream — by the
+                # time the no-tool branch runs ``_flush_queued_messages``,
+                # this item is in the queue waiting to be drained.
+                session.queue_message("late arrival", queue_msg_id="q-late")
+            return iter([])
+
+        with (
+            patch.object(session, "_create_stream_with_retry", side_effect=mock_create_stream),
+            patch.object(
+                session,
+                "_stream_response",
+                return_value={"role": "assistant", "content": "ok"},
+            ),
+            patch.object(session, "_full_messages", return_value=[]),
+            patch.object(session, "_update_token_table"),
+            patch.object(session, "_print_status_line"),
+            patch.object(session, "_emit_state"),
+            patch.object(session, "_visible_memory_count", return_value=0),
+            patch("turnstone.core.session.save_message"),
+        ):
+            session.send("first message")
+
+        # Loop continued: a second stream call happened after the
+        # queued message drained into history.  Pre-fix: 1 call.
+        assert stream_calls == 2, (
+            f"expected loop to continue after drain (2 stream calls); got {stream_calls}"
+        )
+        # The queued message landed in history before the second turn.
+        user_texts: list[str] = []
+        for m in session.messages:
+            if m.get("role") != "user":
+                continue
+            content = m.get("content")
+            if isinstance(content, str):
+                user_texts.append(content)
+            elif isinstance(content, list):
+                for part in content:
+                    if isinstance(part, dict) and "text" in part:
+                        user_texts.append(part["text"])
+        assert any("late arrival" in t for t in user_texts), (
+            f"queued message must appear in history; got user texts: {user_texts!r}"
+        )
+
 
 class TestReminderSidechannelIsolation:
     """The side-channel design's load-bearing guarantee: any reader of

turnstone/core/providers/_openai_common.py

@@ -182,7 +182,7 @@ OPENAI_CAPABILITIES: dict[str, ModelCapabilities] = {
 }
 
 # Default for unknown models (local servers: vLLM, llama.cpp, etc.)
-OPENAI_DEFAULT = ModelCapabilities(supports_tool_advisories=False)
+OPENAI_DEFAULT = ModelCapabilities()
 
 
 def lookup_openai_capabilities(model: str) -> ModelCapabilities:

turnstone/core/providers/_protocol.py

@@ -86,7 +86,6 @@ class ModelCapabilities:
     supports_web_search: bool = False
     supports_tool_search: bool = False
     supports_vision: bool = False
-    supports_tool_advisories: bool = True
     thinking_display: str = ""  # "summarized" for models that omit thinking by default
 
 

turnstone/core/session.py

@@ -2475,7 +2475,12 @@ class ChatSession:
                         threading.Thread(target=self._generate_title, daemon=True).start()
                     # Flush any queued messages that weren't injected
                     # (no tool calls → no advisory seam to inject at).
-                    self._flush_queued_messages()
+                    # If anything drained, the model hasn't seen those
+                    # messages yet — keep the loop alive so it gets a
+                    # turn over the extended history rather than
+                    # orphaning them until the next user send.
+                    if self._flush_queued_messages():
+                        continue
                     self._emit_state("idle")
                     # Dispatch any pending watch results (chains into
                     # a new send() within the same worker thread).
@@ -3863,7 +3868,7 @@ class ChatSession:
             )
         return resolved
 
-    def _flush_queued_messages(self) -> None:
+    def _flush_queued_messages(self) -> bool:
         """Drain queued messages.
 
         Items without attachments are combined into a single user turn
@@ -3871,6 +3876,8 @@ class ChatSession:
         poorly.  Items with attachments flush as separate multipart user
         turns (combining text+files across distinct queued sends would
         misrepresent ordering).
+
+        Returns ``True`` when any items drained, ``False`` otherwise.
         """
         from turnstone.core.tool_advisory import PRIORITY_IMPORTANT
 
@@ -3879,7 +3886,7 @@ class ChatSession:
             items = list(self._queued_messages.items())
             self._queued_messages.clear()
         if not items:
-            return
+            return False
 
         # Collapse contiguous attachment-free items into one combined text
         # to preserve the prior behaviour; flush attachment-bearing items
@@ -3905,6 +3912,7 @@ class ChatSession:
             else:
                 text_run.append((cleaned, priority))
         _flush_text_run()
+        return True
 
     def _collect_advisories(
         self,
@@ -3934,20 +3942,6 @@ class ChatSession:
         """
         from turnstone.core.tool_advisory import GuardAdvisory, UserInterjection
 
-        caps = self._get_capabilities()
-
-        # When the model doesn't support advisory tags, still drain queued
-        # messages so they aren't silently orphaned — flush them as regular
-        # user messages instead. Metacognitive tool advisories (tool_error
-        # / repeat) are dropped silently: the model wouldn't reliably parse
-        # them anyway, and the user-channel nudges still fire on the next
-        # user turn.
-        if not caps.supports_tool_advisories:
-            if is_last_in_batch:
-                self._pending_tool_advisories.clear()
-                self._flush_queued_messages()
-            return [], []
-
         persistent: list[ToolAdvisory] = []
         metacog_reminders: list[dict[str, str]] = []