fix(chat): prevent duplicate bubbles for conflicting provider payloads

Record pending-turn proof under caller-owned session locks during normal
runs and credential recovery. Preserve both conflicting provider payloads
for replay, but display one sidecar-owned bubble when row identity and
visible content are unambiguous.

Assisted-by: Hermes Agent:gpt-6-sol
Assisted-by: Codex:gpt-6-luna

api/models.py

@@ -11537,6 +11537,81 @@ def reconciled_state_db_messages_for_session(
         truncation_boundary=getattr(session, "truncation_boundary", None),
         incoming_provenance="state_db",
     )
+    if not prefer_context:
+        # Keep the append-only merge's conflicting provider payloads in model
+        # context. For display only, a marked native-image mirror can share the
+        # existing sidecar bubble when its durable identity, exact timestamp,
+        # and exact visible user content all agree.
+        sidecar_row_id_counts = collections.Counter(
+            row_id
+            for message in local_messages
+            if (row_id := _state_db_row_identity(message)) is not None
+        )
+        state_row_id_counts = collections.Counter(
+            row_id
+            for message in state_messages or ()
+            if (row_id := _state_db_row_identity(message)) is not None
+        )
+
+        display_conflicts = {}
+        for incoming in state_messages or ():
+            if (
+                not isinstance(incoming, dict)
+                or incoming.get(_WEBUI_UNMATCHED_NATIVE_IMAGE_MIRROR_FIELD) is not True
+            ):
+                continue
+            row_id, row_id_valid = _state_db_row_identity_details(incoming)
+            timestamp, timestamp_valid = _message_exact_timestamp_details(incoming)
+            content = incoming.get("content")
+            if (
+                not row_id_valid
+                or row_id is None
+                or sidecar_row_id_counts[row_id] != 1
+                or state_row_id_counts[row_id] != 1
+                or not timestamp_valid
+                or timestamp is None
+                or str(incoming.get("role") or "").lower() != "user"
+                or not isinstance(content, str)
+            ):
+                continue
+            incoming_api_content = _session_message_api_content_key(incoming)
+            if incoming_api_content is None:
+                continue
+            sidecar_owner = next((
+                message for message in local_messages
+                if isinstance(message, dict)
+                and _state_db_row_identity_details(message) == (row_id, True)
+                and _message_identity_compatible(message, incoming)
+                and message.get("content") == content
+                and _message_exact_timestamp_details(message) == (timestamp, True)
+                and _session_message_api_content_key(message)
+                not in (None, incoming_api_content)
+            ), None)
+            if sidecar_owner is not None:
+                display_conflicts[(row_id, timestamp, content)] = (
+                    incoming,
+                    sidecar_owner,
+                )
+
+        if display_conflicts:
+            visible_messages = []
+            for message in reconciled_messages:
+                if isinstance(message, dict) and isinstance(message.get("content"), str):
+                    row_id, row_id_valid = _state_db_row_identity_details(message)
+                    timestamp, timestamp_valid = _message_exact_timestamp_details(message)
+                    conflict = (
+                        display_conflicts.get((row_id, timestamp, message["content"]))
+                        if row_id_valid and row_id is not None and timestamp_valid
+                        else None
+                    )
+                    if (
+                        conflict is not None
+                        and message is not conflict[1]
+                        and _message_identity_compatible(message, conflict[0])
+                    ):
+                        continue
+                visible_messages.append(message)
+            reconciled_messages = visible_messages
     return _state_db_session_messages_result(
         reconciled_messages,
         state_revision,

api/streaming.py

@@ -9137,10 +9137,42 @@ def _add_supported_run_conversation_kwarg(callable_obj, kwargs, name, value):
     return False
 
 
+def _register_pending_user_timestamp_identity(
+    callable_obj, session, persist_user_timestamp,
+):
+    """Persist the pending-turn proof while the caller owns the session lock."""
+    if session is None:
+        return
+    try:
+        timestamp_parameter = inspect.signature(callable_obj).parameters.get(
+            "persist_user_timestamp"
+        )
+    except (TypeError, ValueError):
+        timestamp_parameter = None
+    has_timestamp_contract = (
+        timestamp_parameter is not None
+        and timestamp_parameter.kind is not inspect.Parameter.POSITIONAL_ONLY
+    )
+    identity = (
+        _validated_webui_pending_user_timestamp_identity(
+            session,
+            (getattr(session, "active_stream_id", None), persist_user_timestamp),
+        )
+        if has_timestamp_contract
+        else None
+    )
+    previous = getattr(session, "_webui_pending_user_timestamp_identity", None)
+    session._webui_pending_user_timestamp_identity = identity
+    session_id = getattr(session, "session_id", None)
+    save = getattr(session, "save", None)
+    if callable(save) and session_id and previous != identity:
+        # Persist before run_conversation can write this turn to state.db.
+        save(touch_updated_at=False, skip_index=True)
+
+
 def _build_run_conversation_kwargs(
     callable_obj,
     *,
-    session=None,
     user_message,
     system_message,
     conversation_history,
@@ -9149,7 +9181,7 @@ def _build_run_conversation_kwargs(
     persist_user_message,
     persist_user_timestamp,
 ):
-    """Build one rolling-compatible Agent invocation contract.
+    """Build one rolling-compatible Agent invocation contract without mutation.
 
     ``persist_user_timestamp`` is signature-gated (#6935): an older
     hermes-agent whose ``run_conversation()`` predates the kwarg must not
@@ -9174,45 +9206,6 @@ def _build_run_conversation_kwargs(
         "conversation_history_revision",
         conversation_history_revision,
     )
-    if session is not None:
-        try:
-            timestamp_parameter = inspect.signature(callable_obj).parameters.get(
-                "persist_user_timestamp"
-            )
-        except (TypeError, ValueError):
-            timestamp_parameter = None
-        has_timestamp_contract = (
-            timestamp_parameter is not None
-            and timestamp_parameter.kind is not inspect.Parameter.POSITIONAL_ONLY
-        )
-        session_id = getattr(session, "session_id", None)
-        save = getattr(session, "save", None)
-        lock = (
-            _get_session_agent_lock(session_id)
-            if callable(save) and session_id
-            else contextlib.nullcontext()
-        )
-        with lock:
-            candidate = (
-                _validated_webui_pending_user_timestamp_identity(
-                    session,
-                    (
-                        getattr(session, "active_stream_id", None),
-                        kwargs.get("persist_user_timestamp"),
-                    ),
-                )
-                if has_timestamp_contract
-                else None
-            )
-            # A retry using an older/opaque Agent must revoke earlier proof;
-            # row identity cannot be carried across a changed invocation contract.
-            identity = candidate
-            previous = getattr(session, "_webui_pending_user_timestamp_identity", None)
-            session._webui_pending_user_timestamp_identity = identity
-            if callable(save) and session_id and previous != identity:
-                # Persist the proof while holding the session mutation lock,
-                # before run_conversation can write this turn to state.db.
-                save(touch_updated_at=False, skip_index=True)
     return kwargs
 
 
@@ -11763,9 +11756,13 @@ def _run_agent_streaming(
                 _agent_msg_text = "\n\n".join([*_process_notifications, msg_text]).strip()
             user_message = _build_native_multimodal_message(workspace_ctx, _agent_msg_text, attachments, workspace, cfg=_cfg, active_provider=(resolved_provider or ""), active_model=(resolved_model or ""), requested_provider=(_session_requested_provider or ""), profile=(getattr(s, "profile", None) or Path(_profile_home)))
             _persistent_state_before = _persistent_state_snapshot(_profile_home)
+            with _agent_lock:
+                _persist_user_timestamp = getattr(s, 'pending_started_at', None)
+                _register_pending_user_timestamp_identity(
+                    agent.run_conversation, s, _persist_user_timestamp
+                )
             _run_conversation_kwargs = _build_run_conversation_kwargs(
                 agent.run_conversation,
-                session=s,
                 user_message=user_message,
                 system_message=workspace_system_msg,
                 conversation_history=_sanitize_messages_for_agent(
@@ -11779,7 +11776,7 @@ def _run_agent_streaming(
                 conversation_history_revision=_conversation_history_revision,
                 task_id=session_id,
                 persist_user_message=msg_text,
-                persist_user_timestamp=getattr(s, 'pending_started_at', None),
+                persist_user_timestamp=_persist_user_timestamp,
             )
             # Only pass moa_config when a /moa override is actually active, so a
             # normal send never trips a TypeError on an older hermes-agent whose
@@ -12343,9 +12340,17 @@ def _run_agent_streaming(
                                     _heal_context_messages,
                                     _heal_conversation_history_revision,
                                 ) = _refresh_context_and_revision_from_state_db()
+                                _heal_persist_user_timestamp = getattr(
+                                    s, 'pending_started_at', None
+                                )
+                                # The returned-error path already owns _agent_lock.
+                                _register_pending_user_timestamp_identity(
+                                    agent.run_conversation,
+                                    s,
+                                    _heal_persist_user_timestamp,
+                                )
                                 _heal_kwargs = _build_run_conversation_kwargs(
                                     agent.run_conversation,
-                                    session=s,
                                     user_message=user_message,
                                     system_message=workspace_system_msg,
                                     conversation_history=_sanitize_messages_for_agent(
@@ -12361,7 +12366,7 @@ def _run_agent_streaming(
                                     ),
                                     task_id=session_id,
                                     persist_user_message=msg_text,
-                                    persist_user_timestamp=getattr(s, 'pending_started_at', None),
+                                    persist_user_timestamp=_heal_persist_user_timestamp,
                                 )
                                 if moa_config is not None:
                                     _heal_kwargs["moa_config"] = moa_config
@@ -13695,9 +13700,17 @@ def _run_agent_streaming(
                             _heal_context_messages,
                             _heal_conversation_history_revision,
                         ) = _refresh_context_and_revision_from_state_db()
+                        with _agent_lock:
+                            _heal_persist_user_timestamp = getattr(
+                                s, 'pending_started_at', None
+                            )
+                            _register_pending_user_timestamp_identity(
+                                _heal_agent.run_conversation,
+                                s,
+                                _heal_persist_user_timestamp,
+                            )
                         _heal_kwargs2 = _build_run_conversation_kwargs(
                             _heal_agent.run_conversation,
-                            session=s,
                             user_message=user_message,
                             system_message=workspace_system_msg,
                             conversation_history=_sanitize_messages_for_agent(
@@ -13713,7 +13726,7 @@ def _run_agent_streaming(
                             ),
                             task_id=session_id,
                             persist_user_message=msg_text,
-                            persist_user_timestamp=getattr(s, 'pending_started_at', None),
+                            persist_user_timestamp=_heal_persist_user_timestamp,
                         )
                         if moa_config is not None:
                             _heal_kwargs2["moa_config"] = moa_config

docs/rfcs/webui-run-state-consistency-contract.md

@@ -187,10 +187,13 @@ and 5; it does not mark every run-state boundary implemented.
    Match settled native-image scalar projections only with trusted turn and
    durable-row identity, never the marker alone. A durable row ID proves row
    identity, not provider-payload freshness: when the sidecar and state.db have
-   conflicting nonempty `api_content`, preserve both versions for display and
+   conflicting nonempty `api_content`, preserve both versions for model-context
-   replay without mutating either. Fill a missing payload from the other copy;
+   replay without mutating either. For visible display, a marked mirror may
-   repeated reconciliation must remain bounded and idempotent. Preserve
+   share the existing sidecar bubble only when its valid durable row ID, exact
-   genuinely distinct or ambiguous rows when matching settled projections.
+   timestamp, and exact visible user content match; keep the sidecar-owned row
+   and its display metadata. Distinct row IDs, ambiguous or invalid identities,
+   and different visible user text remain separate. Fill a missing payload from
+   the other copy; repeated reconciliation must remain bounded and idempotent.
    Agent state.db alone cannot restore the original attachment if the WebUI
    sidecar is lost.
    Visible interim assistant progress must remain visible timeline content; a

tests/test_compression_snapshot_revision.py

@@ -1068,6 +1068,7 @@ def test_auth_self_heal_refreshes_revision_after_first_agent_persists_user(
         messages=prior_messages,
         context_messages=prior_messages,
     )
+    _session.pending_user_source = "webui"
     revisions = []
 
     class AuthThenRecoverAgent:
@@ -1080,8 +1081,14 @@ def test_auth_self_heal_refreshes_revision_after_first_agent_persists_user(
             self._last_error = None
             self.stream_delta_callback = _kwargs.get("stream_delta_callback")
 
-        def run_conversation(self, **kwargs):
+        def run_conversation(self, persist_user_timestamp=None, **kwargs):
             type(self).runs += 1
+            assert persist_user_timestamp == 10.0
+            persisted = Session.load(sid)
+            assert persisted._webui_pending_user_timestamp_identity == (
+                stream_id,
+                persist_user_timestamp,
+            )
             revisions.append(kwargs["conversation_history_revision"])
             history = list(kwargs.get("conversation_history") or [])
             if type(self).runs == 1:

tests/test_native_image_turn_display_context.py

@@ -236,6 +236,10 @@ def test_get_session_keeps_pending_agent_projection_private_but_in_context(
     import api.models as models
     import api.session_ops
     from api import routes
+    from api.streaming import (
+        _build_run_conversation_kwargs,
+        _register_pending_user_timestamp_identity,
+    )
 
     timestamp = time.time()
     stream_id = "pending-webui-stream"
@@ -304,11 +308,10 @@ def test_get_session_keeps_pending_agent_projection_private_but_in_context(
                 ],
             )
 
-    from api.streaming import _build_run_conversation_kwargs
+    with api.config._get_session_agent_lock(session_id):
-
+        _register_pending_user_timestamp_identity(agent_run, session, timestamp)
     run_kwargs = _build_run_conversation_kwargs(
         agent_run,
-        session=session,
         user_message=prompt,
         system_message="system",
         conversation_history=[],
@@ -434,7 +437,11 @@ def test_get_session_keeps_pending_agent_projection_private_but_in_context(
 
 def test_pending_row_identity_requires_timestamp_handoff_to_active_agent():
     from api.models import _suppress_native_image_display_mirrors
-    from api.streaming import _build_run_conversation_kwargs
+    from api.streaming import (
+        _build_run_conversation_kwargs,
+        _register_pending_user_timestamp_identity,
+        _get_session_agent_lock,
+    )
 
     def modern_run(user_message, persist_user_timestamp=None):
         return None
@@ -453,7 +460,6 @@ def test_pending_row_identity_requires_timestamp_handoff_to_active_agent():
         pending_user_source="webui",
     )
     kwargs = {
-        "session": session,
         "user_message": "prompt",
         "system_message": "system",
         "conversation_history": [],
@@ -462,11 +468,15 @@ def test_pending_row_identity_requires_timestamp_handoff_to_active_agent():
         "persist_user_message": "prompt",
         "persist_user_timestamp": timestamp,
     }
+    with _get_session_agent_lock("session"):
+        _register_pending_user_timestamp_identity(modern_run, session, timestamp)
     modern = _build_run_conversation_kwargs(modern_run, **kwargs)
     assert modern["persist_user_timestamp"] == timestamp
     assert session._webui_pending_user_timestamp_identity == (
         "stream-identity", timestamp,
     )
+    with _get_session_agent_lock("session"):
+        _register_pending_user_timestamp_identity(modern_run, session, timestamp)
     rebuilt = _build_run_conversation_kwargs(modern_run, **kwargs)
     assert rebuilt["persist_user_timestamp"] == timestamp
     assert session._webui_pending_user_timestamp_identity == (
@@ -474,6 +484,8 @@ def test_pending_row_identity_requires_timestamp_handoff_to_active_agent():
     )
     # Credential-heal fallback can replace the callable mid-turn. Proof from
     # the first invocation must not authorize rows from an older Agent.
+    with _get_session_agent_lock("session"):
+        _register_pending_user_timestamp_identity(legacy_run, session, timestamp)
     fallback = _build_run_conversation_kwargs(legacy_run, **kwargs)
     assert "persist_user_timestamp" not in fallback
     assert session._webui_pending_user_timestamp_identity is None
@@ -484,14 +496,18 @@ def test_pending_row_identity_requires_timestamp_handoff_to_active_agent():
         pending_user_message="prompt",
         pending_user_source="webui",
     )
-    legacy = _build_run_conversation_kwargs(
+    with _get_session_agent_lock("session"):
-        legacy_run, **{**kwargs, "session": legacy_session}
+        _register_pending_user_timestamp_identity(
-    )
+            legacy_run, legacy_session, timestamp
+        )
+    legacy = _build_run_conversation_kwargs(legacy_run, **kwargs)
     assert "persist_user_timestamp" not in legacy
     assert legacy_session._webui_pending_user_timestamp_identity is None
-    opaque = _build_run_conversation_kwargs(
+    with _get_session_agent_lock("session"):
-        opaque_run, **{**kwargs, "session": legacy_session}
+        _register_pending_user_timestamp_identity(
-    )
+            opaque_run, legacy_session, timestamp
+        )
+    opaque = _build_run_conversation_kwargs(opaque_run, **kwargs)
     assert "persist_user_timestamp" in opaque
     assert legacy_session._webui_pending_user_timestamp_identity is None
     row = {"role": "user", "content": "Agent-only memory", "timestamp": timestamp}
@@ -1204,12 +1220,18 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery(
         if message.get("_active_turn_token") == identity["token"]
     )
     mirror = _durable_agent_content(context_user["content"])
+    display_attachments = [{
+        "name": "sidecar-owned.png",
+        "mime": "image/png",
+        "is_image": True,
+    }]
     sidecar_mirror_row = {
         "role": "user",
         "content": mirror,
         "timestamp": timestamp,
         "_state_db_row_id": 42,
         "api_content": sidecar_payload,
+        "attachments": display_attachments,
     }
     session.messages.append(dict(sidecar_mirror_row))
     session.context_messages.append(dict(sidecar_mirror_row))
@@ -1243,10 +1265,9 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery(
             message for message in context
             if message.get("_state_db_row_id") == 42
         ]
-        assert len(display_row_42) == 2
+        assert len(display_row_42) == 1
-        assert {message["api_content"] for message in display_row_42} == {
+        assert display_row_42[0]["api_content"] == sidecar_payload
-            sidecar_payload, state_payload,
+        assert display_row_42[0]["attachments"] == display_attachments
-        }
         assert len(context_row_42) == 2
         assert {message["api_content"] for message in context_row_42} == {
             sidecar_payload, state_payload,
@@ -1254,8 +1275,9 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery(
 
         public = public_session_projection({"messages": display})["messages"]
         public_mirrors = [message for message in public if message.get("content") == mirror]
-        assert len(public_mirrors) == 2
+        assert len(public_mirrors) == 1
         assert all("api_content" not in message for message in public_mirrors)
+        assert public_mirrors[0]["attachments"] == display_attachments
         display_image_turn = next(
             message for message in display
             if message.get("_active_turn_token") == identity["token"]
@@ -1277,14 +1299,8 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery(
             message for message in session.messages
             if message.get("_state_db_row_id") == 42
         ]
-        if first_public is None:
+        assert len(live_row_42) == 1
-            assert len(live_row_42) == 1
+        assert live_row_42[0]["api_content"] == sidecar_payload
-            assert live_row_42[0]["api_content"] == sidecar_payload
-        else:
-            assert len(live_row_42) == 2
-            assert {message["api_content"] for message in live_row_42} == {
-                sidecar_payload, state_payload,
-            }
         replay_image_turn = next(
             message for message in replay
             if isinstance(message.get("content"), list)
@@ -1306,6 +1322,106 @@ def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery(
         session.context_messages = context
 
 
+def test_native_image_payload_conflict_with_distinct_visible_text_stays_separate():
+    import api.models as models
+
+    timestamp = 865.0
+    session, identity, _ = _settle_image_turn(timestamp=timestamp, agent_row_id=41)
+    context_user = next(
+        message for message in session.context_messages
+        if message.get("_active_turn_token") == identity["token"]
+    )
+    mirror = _durable_agent_content(context_user["content"])
+    sidecar_row = {
+        "role": "user",
+        "content": mirror,
+        "timestamp": timestamp,
+        "_state_db_row_id": 42,
+        "api_content": "SIDECAR-PAYLOAD",
+    }
+    session.messages.append(sidecar_row)
+    state_row = {
+        "role": "user",
+        "content": "A separate visible submission",
+        "timestamp": timestamp,
+        "_state_db_row_id": 42,
+        "api_content": "STATE-DB-PAYLOAD",
+        "_webui_unmatched_native_image_mirror": True,
+    }
+
+    display = models.reconciled_state_db_messages_for_session(
+        session,
+        state_messages=[state_row],
+    )
+    same_row = [
+        message for message in display
+        if message.get("_state_db_row_id") == 42
+    ]
+    assert {message["content"] for message in same_row} == {
+        mirror,
+        "A separate visible submission",
+    }
+
+
+def test_marked_native_image_conflict_keeps_ambiguous_sidecar_row_id_visible():
+    import api.models as models
+
+    timestamp = 867.0
+    session, identity, _ = _settle_image_turn(timestamp=timestamp, agent_row_id=41)
+    context_user = next(
+        message for message in session.context_messages
+        if message.get("_active_turn_token") == identity["token"]
+    )
+    mirror = _durable_agent_content(context_user["content"])
+    sidecar_rows = [
+        {
+            "role": "user",
+            "content": mirror,
+            "timestamp": timestamp,
+            "_state_db_row_id": 42,
+            "api_content": "SIDECAR-PAYLOAD-ONE",
+            "attachments": [{"name": "first.png"}],
+        },
+        {
+            "role": "user",
+            "content": mirror,
+            "timestamp": timestamp,
+            "_state_db_row_id": 42,
+            "api_content": "SIDECAR-PAYLOAD-TWO",
+            "attachments": [{"name": "second.png"}],
+        },
+    ]
+    session.messages.extend(sidecar_rows)
+    state_row = {
+        "role": "user",
+        "content": mirror,
+        "timestamp": timestamp,
+        "_state_db_row_id": 42,
+        "api_content": "STATE-DB-PAYLOAD",
+    }
+    marked = models._suppress_native_image_display_mirrors(session, [state_row])
+    assert marked[0]["_webui_unmatched_native_image_mirror"] is True
+
+    display = models.reconciled_state_db_messages_for_session(
+        session,
+        state_messages=[state_row],
+    )
+    same_row = [
+        message for message in display
+        if message.get("_state_db_row_id") == 42
+    ]
+    assert len(same_row) == 3
+    assert {message["api_content"] for message in same_row} == {
+        "SIDECAR-PAYLOAD-ONE",
+        "SIDECAR-PAYLOAD-TWO",
+        "STATE-DB-PAYLOAD",
+    }
+    assert [message.get("attachments") for message in same_row[:2]] == [
+        [{"name": "first.png"}],
+        [{"name": "second.png"}],
+    ]
+
+
 def test_marked_native_image_rows_with_distinct_ids_survive_recovery():
     import api.models as models