fix(chat): bound conflicting recovered image rows

Keep conflicting provider payloads distinct, while deduplicating the durable row across repeated reconciliation.

Assisted-by: Hermes Agent:gpt-6-sol
Assisted-by: Codex:gpt-6-luna
Assisted-by: Claude Code:claude-opus-5-5

tests/test_native_image_turn_display_context.py

@@ -799,6 +799,76 @@ def test_marked_native_image_mirror_repairs_malformed_sidecar_once():
         assert replay_rows[0]["api_content"] == api_content
 
 
+def test_marked_native_image_mirror_conflict_stays_bounded_across_recovery():
+    import api.models as models
+
+    timestamp = 860.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"])
+    session.messages.append({
+        "role": "user",
+        "content": mirror,
+        "timestamp": timestamp,
+        "_state_db_row_id": 42,
+        "api_content": "OLD-PROVIDER-BYTES",
+    })
+    state_row = {
+        "role": "user",
+        "content": mirror,
+        "timestamp": timestamp,
+        "_state_db_row_id": 42,
+        "api_content": "NEW-PROVIDER-BYTES",
+    }
+    marked = models._suppress_native_image_display_mirrors(session, [state_row])
+    assert marked[0]["_webui_unmatched_native_image_mirror"] is True
+
+    first_public = None
+    first_replay = None
+    for _ in range(4):
+        display = models.reconciled_state_db_messages_for_session(
+            session,
+            state_messages=[state_row],
+        )
+        context = models.reconciled_state_db_messages_for_session(
+            session,
+            prefer_context=True,
+            state_messages=[state_row],
+        )
+        display_row_42 = [
+            message for message in display
+            if message.get("_state_db_row_id") == 42
+        ]
+        context_row_42 = [
+            message for message in context
+            if message.get("_state_db_row_id") == 42
+        ]
+        assert len(display_row_42) == 2
+        assert {message["api_content"] for message in display_row_42} == {
+            "OLD-PROVIDER-BYTES",
+            "NEW-PROVIDER-BYTES",
+        }
+        assert len(context_row_42) == 1
+        assert context_row_42[0]["api_content"] == "NEW-PROVIDER-BYTES"
+
+        public = public_session_projection({"messages": display})["messages"]
+        replay = _sanitize_messages_for_agent(context)
+        if first_public is None:
+            first_public = public
+            first_replay = replay
+        else:
+            assert public == first_public
+            assert replay == first_replay
+        session.messages = display
+        session.context_messages = context
+
+
 def test_unlinked_state_db_image_projection_uses_existing_reconciliation():
     import api.models as models