fix(chat): project payload conflicts in session reloads

Apply the same display-only conflict projection to full and paginated
session reads before cache and window selection. Keep both provider payloads
for model replay while retaining a single visible sidecar bubble.

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

api/models.py

@@ -10717,6 +10717,86 @@ def merge_session_messages_append_only(
         _STRUCTURED_IDENTITY_MEMO.reset(token)
 
 
+def _project_native_image_payload_conflicts_for_display(
+    sidecar_messages,
+    state_messages,
+    merged_messages,
+):
+    """Project a proven same-row native-image conflict onto its sidecar bubble."""
+    sidecar_messages = list(sidecar_messages or [])
+    state_messages = list(state_messages or [])
+    if not sidecar_messages or not state_messages:
+        return merged_messages
+
+    sidecar_row_id_counts = collections.Counter(
+        row_id
+        for message in sidecar_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
+        if (row_id := _state_db_row_identity(message)) is not None
+    )
+    display_conflicts = {}
+    for incoming in state_messages:
+        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 sidecar_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 not display_conflicts:
+        return merged_messages
+    visible_messages = []
+    for message in merged_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)
+    return visible_messages
+
+
 def _merge_session_messages_append_only_impl(
     sidecar_messages: list,
     state_messages: list,
@@ -11538,80 +11618,11 @@ def reconciled_state_db_messages_for_session(
         incoming_provenance="state_db",
     )
     if not prefer_context:
-        # Keep the append-only merge's conflicting provider payloads in model
+        reconciled_messages = _project_native_image_payload_conflicts_for_display(
-        # context. For display only, a marked native-image mirror can share the
+            local_messages,
-        # existing sidecar bubble when its durable identity, exact timestamp,
+            state_messages,
-        # and exact visible user content all agree.
+            reconciled_messages,
-        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/routes.py

@@ -9304,6 +9304,7 @@ def _limited_webui_messages_for_display_with_sidecar(
     )
     if not state_db_messages:
         return sidecar_messages
+
     # NOTE: do not short-circuit to the sidecar when state.db has no strictly
     # newer rows. A state.db row whose timestamp is at-or-before the sidecar's
     # newest (recovery / edited-in-place / missing-timestamp cases) is still
@@ -9364,6 +9365,11 @@ def _limited_webui_messages_for_display_with_sidecar(
         truncation_boundary=getattr(session, "truncation_boundary", None),
         incoming_provenance="state_db",
     )
+    merged = _project_native_image_payload_conflicts_for_display(
+        sidecar_messages,
+        state_db_messages,
+        merged,
+    )
     if cache_key is not None:
         _state_key = cache_key[4]
         _streaming_key = (
@@ -10580,6 +10586,7 @@ from api.models import (
     get_state_db_session_message_keys_before_timestamp,
     get_state_db_session_summary,
     merge_session_messages_append_only,
+    _project_native_image_payload_conflicts_for_display,
     _suppress_native_image_display_mirrors,
     _reconcile_api_content_sidecars,
     _enrich_sidebar_lineage_metadata,
@@ -13325,12 +13332,18 @@ def _handle_session_get(handler, parsed) -> bool:
                     s,
                     state_db_messages,
                 )
+                sidecar_messages = _webui_sidecar_lineage_messages_for_display(s)
                 _all_msgs = merge_session_messages_append_only(
-                    _webui_sidecar_lineage_messages_for_display(s),
+                    sidecar_messages,
                     state_db_messages,
                     truncation_watermark=getattr(s, "truncation_watermark", None),
                     truncation_boundary=getattr(s, "truncation_boundary", None),
                 )
+                _all_msgs = _project_native_image_payload_conflicts_for_display(
+                    sidecar_messages,
+                    state_db_messages,
+                    _all_msgs,
+                )
                 _all_msgs = _merged_webui_lineage_messages_for_display(s, _all_msgs)
         else:
             if is_messaging_session and cli_messages:

tests/test_native_image_turn_display_context.py

@@ -227,6 +227,149 @@ def _write_state_db(path, session_id, rows):
         )
 
 
+@pytest.mark.parametrize("msg_limit", [None, 5])
+def test_get_session_projects_marked_payload_conflict_in_full_and_limited_paths(
+    monkeypatch, tmp_path, msg_limit,
+):
+    import api.config
+    import api.models as models
+    import api.session_ops
+    from api import routes
+
+    session_id = f"native-image-route-{msg_limit}"
+    timestamp = 880.0
+    seed, identity, _ = _settle_image_turn(
+        session_id=session_id,
+        timestamp=timestamp,
+        agent_row_id=41,
+    )
+    context_user = next(
+        message for message in seed.context_messages
+        if message.get("_active_turn_token") == identity["token"]
+    )
+    mirror = _durable_agent_content(context_user["content"])
+    sidecar_payload = "SIDE-CAR-PROVIDER-PAYLOAD"
+    state_payload = "STATE-DB-PROVIDER-PAYLOAD"
+    sidecar_attachments = [{
+        "name": "sidecar-owned.png",
+        "mime": "image/png",
+        "is_image": True,
+    }]
+    sidecar_mirror = {
+        "role": "user",
+        "content": mirror,
+        "timestamp": timestamp,
+        "_state_db_row_id": 1,
+        "api_content": sidecar_payload,
+        "attachments": sidecar_attachments,
+    }
+    seed.messages.append(dict(sidecar_mirror))
+    seed.context_messages.append(dict(sidecar_mirror))
+
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    db_path = tmp_path / "state.db"
+    _write_state_db(db_path, session_id, [(mirror, timestamp, state_payload)])
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json")
+    monkeypatch.setattr(routes, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db_path)
+    session = models.Session(
+        session_id=session_id,
+        workspace="/fixture",
+        model="fixture-model",
+        context_length=128_000,
+        messages=seed.messages,
+        context_messages=seed.context_messages,
+        source_tag="webui",
+        session_source="webui",
+    )
+    session.save(skip_index=True)
+    with models.LOCK:
+        models.SESSIONS.pop(session_id, None)
+
+    monkeypatch.setattr(routes, "_session_visible_to_active_profile", lambda *_: True)
+    monkeypatch.setattr(routes, "_active_stream_ids", lambda: set())
+    monkeypatch.setattr(routes, "_lookup_cli_session_metadata", lambda *_: {})
+    monkeypatch.setattr(routes, "find_run_summary", lambda *_: None)
+    monkeypatch.setattr(api.config, "load_settings", lambda: {"api_redact_enabled": False})
+    monkeypatch.setattr(api.session_ops, "regeneration_state", lambda _session: ([], []))
+    monkeypatch.setattr(
+        api.session_ops,
+        "regeneration_authority",
+        lambda *_args, **_kwargs: None,
+    )
+    response = {}
+    monkeypatch.setattr(
+        routes,
+        "j",
+        lambda _handler, payload, status=200, **_kwargs: (
+            response.update(payload=payload, status=status) or payload
+        ),
+    )
+    query = f"session_id={session_id}&resolve_model=0"
+    if msg_limit is not None:
+        query += f"&msg_limit={msg_limit}"
+    routes._handle_session_get(None, SimpleNamespace(path="/api/session", query=query))
+
+    assert response["status"] == 200
+    public_session = response["payload"]["session"]
+    public_messages = public_session["messages"]
+    mirror_rows = [message for message in public_messages if message.get("content") == mirror]
+    assert len(mirror_rows) == 1, f"expected one visible mirror, found {len(mirror_rows)}"
+    assert all("api_content" not in message for message in mirror_rows)
+    assert public_session["message_count"] == 3
+    assert mirror_rows[0]["attachments"] == sidecar_attachments
+
+    state_rows = models.get_state_db_session_messages(session_id)
+    replay_context = models.reconciled_state_db_messages_for_session(
+        session,
+        prefer_context=True,
+        state_messages=state_rows,
+    )
+    replay_rows = [
+        message for message in _sanitize_messages_for_agent(replay_context)
+        if message.get("content") == mirror
+    ]
+    assert len(replay_rows) == 2
+    assert {message["api_content"] for message in replay_rows} == {
+        sidecar_payload,
+        state_payload,
+    }
+
+
+def test_limited_conflict_projection_keeps_state_row_when_owner_is_outside_slice():
+    from api import routes
+
+    timestamp = 881.0
+    mirror = "Describe this image [screenshot]"
+    sidecar_slice = [{"role": "user", "content": "Earlier submission", "timestamp": 879.0}]
+    marked_state_row = {
+        "role": "user",
+        "content": mirror,
+        "timestamp": timestamp,
+        "_state_db_row_id": 42,
+        "api_content": "STATE-DB-PROVIDER-PAYLOAD",
+        "_webui_unmatched_native_image_mirror": True,
+    }
+    session = SimpleNamespace(
+        session_id="limited-conflict-owner-outside-page",
+        messages=sidecar_slice,
+        context_messages=[],
+    )
+    merged = routes._limited_webui_messages_for_display_with_sidecar(
+        session,
+        sidecar_slice,
+        [marked_state_row],
+        state_db_signature=None,
+        msg_before=1,
+    )
+    assert any(message.get("content") == "Earlier submission" for message in merged)
+    state_rows = [message for message in merged if message.get("content") == mirror]
+    assert len(state_rows) == 1
+    assert state_rows[0]["api_content"] == marked_state_row["api_content"]
+
+
 @pytest.mark.parametrize("msg_limit", [None, 5])
 @pytest.mark.parametrize("attachment_mode", ["native_image", "text_attachment"])
 def test_get_session_keeps_pending_agent_projection_private_but_in_context(