diff --git a/api/models.py b/api/models.py
index 1077cde7..2d0bd03a 100644
--- a/api/models.py
+++ b/api/models.py
@@ -11021,24 +11021,27 @@ def _merge_session_messages_append_only_impl(
                 duplicate_row_id, duplicate_row_id_valid = (
                     _state_db_row_identity_details(duplicate)
                 )
-                same_durable_row = (
+                two_distinct_durable_rows = (
                     row_id_valid
                     and row_id is not None
                     and duplicate_row_id_valid
+                    and duplicate_row_id is not None
+                    and duplicate_row_id != row_id
+                )
+                if not two_distinct_durable_rows:
+                    _merge_session_display_metadata(duplicate, msg)
+                    continue
+                duplicate = merged_by_row_id.get(row_id)
+                duplicate_row_id, duplicate_row_id_valid = (
+                    _state_db_row_identity_details(duplicate)
+                )
+                same_durable_row = (
+                    row_id not in ambiguous_row_ids
+                    and duplicate_row_id_valid
                     and duplicate_row_id == row_id
+                    and _cached_message_key(duplicate, "dedup") == dedup_key
+                    and _row_id_fast_path_allowed(duplicate, msg)
                 )
-                if not same_durable_row and row_id_valid and row_id is not None:
-                    duplicate = merged_by_row_id.get(row_id)
-                    duplicate_row_id, duplicate_row_id_valid = (
-                        _state_db_row_identity_details(duplicate)
-                    )
-                    same_durable_row = (
-                        row_id not in ambiguous_row_ids
-                        and duplicate_row_id_valid
-                        and duplicate_row_id == row_id
-                        and _cached_message_key(duplicate, "dedup") == dedup_key
-                        and _row_id_fast_path_allowed(duplicate, msg)
-                    )
                 if same_durable_row:
                     _merge_session_display_metadata(duplicate, msg)
                     continue
diff --git a/tests/test_native_image_turn_display_context.py b/tests/test_native_image_turn_display_context.py
index a04b145f..155e4708 100644
--- a/tests/test_native_image_turn_display_context.py
+++ b/tests/test_native_image_turn_display_context.py
@@ -4,6 +4,8 @@ import json
 import sqlite3
 from types import SimpleNamespace
 
+import pytest
+
 from api.helpers import public_session_projection
 from api.models import get_state_db_session_messages
 from api.process_event_utils import build_active_turn_token
@@ -926,6 +928,85 @@ def test_marked_native_image_rows_with_distinct_ids_survive_recovery():
         session.context_messages = context
 
 
+@pytest.mark.parametrize(
+    ("row_identity", "copies"),
+    [
+        pytest.param({}, 1, id="idless-row"),
+        pytest.param({}, 2, id="duplicate-idless-rows"),
+        pytest.param({"_state_db_row_id": "x"}, 1, id="malformed-row-id"),
+        pytest.param(
+            {"_row_id": 42, "_state_db_row_id": 43},
+            1,
+            id="conflicting-row-id-aliases",
+        ),
+    ],
+)
+def test_untrusted_native_image_row_identity_deduplicates_stably(
+    row_identity, copies
+):
+    import api.models as models
+
+    timestamp = 880.0
+    session, identity, api_content = _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"])
+    state_rows = [
+        {
+            "role": "user",
+            "content": mirror,
+            "timestamp": timestamp,
+            "api_content": api_content,
+            **row_identity,
+        }
+        for _ in range(copies)
+    ]
+
+    first_public = None
+    first_replay = None
+    for _ in range(4):
+        display = models.reconciled_state_db_messages_for_session(
+            session,
+            state_messages=state_rows,
+        )
+        context = models.reconciled_state_db_messages_for_session(
+            session,
+            prefer_context=True,
+            state_messages=state_rows,
+        )
+        for messages in (display, context):
+            mirrored_rows = [
+                message for message in messages
+                if message.get("content") == mirror
+                and message.get("timestamp") == timestamp
+            ]
+            assert len(mirrored_rows) == 1
+
+        public = public_session_projection({"messages": display})["messages"]
+        public_mirrors = [
+            message for message in public
+            if message.get("content") == mirror
+            and message.get("timestamp") == timestamp
+        ]
+        assert len(public_mirrors) == 1
+        assert "api_content" not in public_mirrors[0]
+        assert all("api_content" not in message for message in public)
+        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