fix(agent): type a synthesized user turn when its row is written

The auto-continue recovery note was typed only after run_conversation
returned, so its row sat untyped for the whole turn — and permanently
when the continuation was itself killed, which is the case it exists
for. persist_user_display_kind stamps the type on the live message
before the crash persist writes it, in the same insert as the content.
The flush also carries display_metadata through, which it was dropping.

agent/conversation_loop.py

@@ -1093,6 +1093,8 @@ def run_conversation(
     stream_callback: Optional[callable] = None,
     persist_user_message: Optional[Any] = None,
     persist_user_timestamp: Optional[float] = None,
+    persist_user_display_kind: Optional[str] = None,
+    persist_user_display_metadata: Optional[Dict[str, Any]] = None,
     moa_config: Optional[dict[str, Any]] = None,
 ) -> Dict[str, Any]:
     """
@@ -1111,6 +1113,13 @@ def run_conversation(
             synthetic prefixes.
         persist_user_timestamp: Optional platform event timestamp to store
             as metadata on that persisted user message.
+        persist_user_display_kind: Optional presentation type for a
+            synthesized user turn (``auto_continue``, ``model_switch``, …).
+            Display-only: transcript surfaces render the row as a timeline
+            event instead of a user bubble, while the model still receives
+            the message unchanged.
+        persist_user_display_metadata: Optional payload for that event
+            (e.g. a delegation's task count).
                 or queuing follow-up prefetch work.
 
     Returns:
@@ -1153,6 +1162,8 @@ def run_conversation(
         stream_callback,
         persist_user_message,
         persist_user_timestamp,
+        persist_user_display_kind=persist_user_display_kind,
+        persist_user_display_metadata=persist_user_display_metadata,
         restore_or_build_system_prompt=_restore_or_build_system_prompt,
         install_safe_stdio=_install_safe_stdio,
         sanitize_surrogates=_sanitize_surrogates,

agent/turn_context.py

@@ -336,6 +336,8 @@ def build_turn_context(
     persist_user_message: Optional[Any],
     persist_user_timestamp: Optional[float] = None,
     *,
+    persist_user_display_kind: Optional[str] = None,
+    persist_user_display_metadata: Optional[Dict[str, Any]] = None,
     restore_or_build_system_prompt,
     install_safe_stdio,
     sanitize_surrogates,
@@ -537,6 +539,19 @@ def build_turn_context(
     # Add the current user message after the prompt/session setup has made
     # close persistence safe. The handoff above preserves any marker already
     # stamped by an earlier close flush.
+    #
+    # A synthesized turn (auto-continue recovery note, delegation completion)
+    # declares how it should READ in a transcript. Stamp that on the live
+    # message so the crash persist below writes the row already typed. Typing
+    # it after the turn instead leaves the row untyped for the whole run — and
+    # forever if the turn crashes — so the raw system note paints as a user
+    # bubble. The model still receives role/content unchanged; the api_messages
+    # build strips both fields from every outgoing copy.
+    if persist_user_display_kind:
+        user_msg["display_kind"] = persist_user_display_kind
+        if persist_user_display_metadata:
+            user_msg["display_metadata"] = persist_user_display_metadata
+
     messages.append(user_msg)
     current_turn_user_idx = len(messages) - 1
     agent._persist_user_message_idx = current_turn_user_idx

run_agent.py

@@ -2112,6 +2112,7 @@ class AIAgent:
                         and not msg.get("_compressed_summary_has_user_turn")
                         else msg.get("display_kind")
                     ),
+                    display_metadata=msg.get("display_metadata"),
                     compression_lock_holder=getattr(
                         self, "_active_compression_lock_holder", None
                     ),
@@ -6842,6 +6843,8 @@ class AIAgent:
         stream_callback: Optional[callable] = None,
         persist_user_message: Optional[Any] = None,
         persist_user_timestamp: Optional[float] = None,
+        persist_user_display_kind: Optional[str] = None,
+        persist_user_display_metadata: Optional[Dict[str, Any]] = None,
         moa_config: Optional[dict[str, Any]] = None,
     ) -> Dict[str, Any]:
         """Forwarder — see ``agent.conversation_loop.run_conversation``."""
@@ -6886,6 +6889,8 @@ class AIAgent:
                     stream_callback,
                     persist_user_message,
                     persist_user_timestamp=persist_user_timestamp,
+                    persist_user_display_kind=persist_user_display_kind,
+                    persist_user_display_metadata=persist_user_display_metadata,
                     moa_config=moa_config,
                 )
             finally:

tests/agent/test_synthetic_turn_display_kind.py

@@ -0,0 +1,106 @@
+"""A synthesized turn's row is typed when it is WRITTEN, not when it ends.
+
+Synthetic user turns (the crash-recovery note, delegation completions) are
+model-facing scaffolding that must read as a timeline event, never as a user
+bubble. The type used to be stamped only after ``run_conversation`` returned,
+which left two holes:
+
+* the row sits untyped for the whole turn, so any surface reading the
+  transcript mid-turn paints the raw ``[System note: …]`` text as if the user
+  had typed it;
+* a turn killed before it concludes never reaches the stamp at all — and the
+  auto-continue turn is precisely the one that gets killed twice, so its note
+  stays a user bubble forever.
+
+``persist_user_display_kind`` moves the typing to turn start, where the
+crash-resilience persist writes it in the same insert as the content.
+"""
+
+from __future__ import annotations
+
+import shutil
+import tempfile
+import types
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from agent.turn_context import build_turn_context
+from hermes_state import SessionDB
+from run_agent import AIAgent
+
+NOTE = "[System note: Your previous turn was interrupted mid-run …]\n\nkeep going"
+
+
+@pytest.fixture()
+def agent_db():
+    tmp = tempfile.mkdtemp(prefix="synthetic_display_kind_")
+    db = SessionDB(Path(tmp) / "state.db")
+    sid = "sess-synthetic"
+    db.create_session(session_id=sid, source="desktop", model="test-model")
+    agent = AIAgent(
+        api_key="test-key",
+        base_url="https://openrouter.ai/api/v1",
+        quiet_mode=True,
+        skip_context_files=True,
+        skip_memory=True,
+        session_db=db,
+        session_id=sid,
+    )
+    agent._session_db_created = True
+    agent._cached_system_prompt = "SYSTEM"
+    agent._skip_mcp_refresh = True
+    try:
+        yield agent, db, sid
+    finally:
+        db.close()
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+def _build(agent, **overrides):
+    kwargs = dict(
+        agent=agent,
+        user_message=NOTE,
+        system_message=None,
+        conversation_history=None,
+        task_id=None,
+        stream_callback=None,
+        persist_user_message=None,
+        restore_or_build_system_prompt=lambda *a, **k: None,
+        install_safe_stdio=lambda: None,
+        sanitize_surrogates=lambda s: s,
+        summarize_user_message_for_log=lambda s: s,
+        set_session_context=lambda _sid: None,
+        set_current_write_origin=lambda _o: None,
+        ra=lambda: types.SimpleNamespace(_set_interrupt=lambda *a, **k: None),
+    )
+    kwargs.update(overrides)
+    with patch("agent.auxiliary_client.set_runtime_main", lambda *a, **k: None):
+        return build_turn_context(**kwargs)
+
+
+def test_row_is_typed_by_the_turn_start_persist(agent_db):
+    agent, db, sid = agent_db
+
+    _build(
+        agent,
+        persist_user_display_kind="auto_continue",
+        persist_user_display_metadata={"attempt": 2},
+    )
+
+    row, = [r for r in db.get_messages_as_conversation(sid) if r["role"] == "user"]
+    # Typed before the turn ran — a crash from here on still reads as an event.
+    assert row["display_kind"] == "auto_continue"
+    assert row["display_metadata"] == {"attempt": 2}
+    # The model's copy is untouched: same role, same content.
+    assert row["content"] == NOTE
+
+
+def test_a_real_user_turn_stays_untyped(agent_db):
+    agent, db, sid = agent_db
+
+    _build(agent, user_message="keep going")
+
+    row, = [r for r in db.get_messages_as_conversation(sid) if r["role"] == "user"]
+    assert row.get("display_kind") is None